Core Message Bus Architecture

Overview

The Core Message Bus is the central nervous system of AICO, enabling modular, event-driven communication between all system components. It implements a publish-subscribe (pub/sub) pattern that allows modules to communicate without direct dependencies, supporting AICO's core principles of modularity, autonomy, and extensibility.

🔒 Security First: All message bus communication is encrypted using CurveZMQ with mandatory authentication. There is no plaintext fallback - the system enforces secure communication or fails completely.

⚠️ CRITICAL: Logging Recursion Prevention - Avoid standard logging within message bus operations to prevent infinite recursion loops.

This architecture document describes the design, implementation, and integration patterns of AICO's central message bus system, which serves as the foundation for inter-module communication and coordination.

Design Principles

The Core Message Bus architecture is built on the following key principles:

1. Loose Coupling

Modules communicate exclusively through the message bus rather than direct method calls, enabling: - Independent development and testing of modules - Ability to replace or upgrade modules without affecting others - Simplified integration of new capabilities

2. Event-Driven Architecture

The system operates on an event-driven paradigm where: - Modules publish events (messages) when state changes occur - Interested modules subscribe to relevant topics - Processing occurs asynchronously and reactively

3. Standardized Communication

All messages follow a consistent envelope structure defined in Protocol Buffers. Conceptually:

message AicoMessage {
  MessageMetadata metadata = 1;
  google.protobuf.Any any_payload = 2; // Domain-specific message
}

message MessageMetadata {
  string message_id = 1;   // UUID
  string timestamp = 2;    // ISO 8601
  string source = 3;       // Originating module
  string message_type = 4; // Topic name from AICOTopics
  string version = 5;      // Schema version
}

4. Topic-Based Routing

Messages are organized in a hierarchical topic structure: - Primary category (e.g., emotion, personality, agency) - Subcategory (e.g., state, expression, goals) - Action/type (e.g., current, update, request)

5. Versioned Message Formats

All message formats are explicitly versioned to enable: - Backward compatibility - Graceful evolution of the system - Support for multiple message format versions simultaneously

Technical Implementation

Message Bus Architecture

The Core Message Bus implements a hybrid broker pattern with the backend service acting as the central message coordinator:

Internal Communication (Backend Modules): - Protocol: ZeroMQ with Protocol Buffers 6.32 - Transport: TCP for all communication (inproc/ipc not used) - Pattern: Pub/Sub with topic hierarchy and wildcard pattern matching - Broker: Backend service runs central ZeroMQ broker on ports 5555 (frontend) and 5556 (backend) - Encryption: Mandatory CurveZMQ with no plaintext fallback

External Communication (Subsystems): - Frontend (Flutter): REST API (Dio client) + WebSocket for streaming, Protocol Buffers 5.0 (wire-compatible) - CLI (Python): Direct ZeroMQ with CurveZMQ encryption, REST API fallback - Studio (React): REST API for admin operations (early development) - Transport: All external clients connect to backend's API Gateway on port 8771

Message Bus Technology

The Core Message Bus uses ZeroMQ with CurveZMQ encryption:

• High-performance: Asynchronous messaging with minimal overhead
• Secure by default: Mandatory CurveZMQ encryption for all communication
• Flexible patterns: Pub/sub with hierarchical topic routing
• Embedded: No external message broker dependencies

Message Format

Protocol Buffers provide: - Binary serialization: Compact, fast encoding/decoding - Strong typing: Compile-time validation and code generation - Versioning: Backward compatibility through schema evolution - Cross-language: Python (6.32), Dart (5.0) - wire-compatible - Production Status: All core messages use protobuf (logs, events, modelservice requests)

Message Validation

Messages are validated through Protocol Buffers' built-in validation: - Compile-time type checking - Runtime schema validation - Required fields enforcement - Automatic versioning support

Topic Hierarchy

The message bus uses a hierarchical topic structure that organizes messages by functional domain and purpose:

Core Domains

IMPORTANT: AICO uses a centralized topic registry (AICOTopics) with slash-based notation for all message bus topics.

ZeroMQ Subscription Behavior

Critical Considerations

1. ZeroMQ uses prefix matching only
2. When you subscribe to a pattern, ZeroMQ converts it to a prefix filter
3. Example: logs/* becomes ZMQ filter logs/

4. This means ZeroMQ will deliver ANY message whose topic starts with that prefix

5. Application-level pattern matching

6. After ZeroMQ delivers messages based on prefix, AICO performs application-level pattern matching
7. This is where wildcard semantics are applied

ZeroMQ Prefix Matching

ZeroMQ uses simple prefix matching (no wildcards):

Pattern	ZMQ Filter	Behavior	Matches
logs/backend	logs/backend	Exact prefix match	logs/backend, logs/backend/main, logs/backend/api
logs/	logs/	Prefix match	All topics starting with logs/
* or **	"" (empty)	Match all	Every message on the bus

Common Subscription Patterns

Use Case	Pattern	ZMQ Filter	Matches
All logs	logs/	logs/	All topics starting with logs/
Backend logs	logs/backend/	logs/backend/	All topics starting with logs/backend/
Specific module	logs/backend/main	logs/backend/main	Topics starting with logs/backend/main
All messages	* or **	"" (empty)	Every message on the bus

Best Practices

1. Use prefix patterns for hierarchical subscriptions
2. Subscribe to logs/ to receive all log messages
3. Subscribe to logs/backend/ to receive all backend logs

4. Be specific with prefixes to avoid unnecessary message delivery

5. Understand ZeroMQ's prefix behavior

6. ZeroMQ delivers ANY message whose topic starts with your filter
7. No application-level filtering is implemented
8. Design topics carefully to leverage prefix matching effectively

Common Pitfalls

1. Expecting wildcard behavior
2. ZeroMQ does NOT support * or ** wildcards
3. logs/* is treated as literal prefix logs/*, not a wildcard

4. Use proper prefixes like logs/ instead

5. Over-subscribing with broad prefixes

6. Subscribing to logs/ delivers ALL log messages
7. This can cause performance issues with high message volume

8. Use specific prefixes when possible

9. Inconsistent topic structure

10. Design hierarchical topics to work well with prefix matching

11. Use consistent separators (slashes) for topic hierarchy

12. emotion/ - Emotion simulation related messages

13. emotion/state/current - Current emotional state
14. emotion/state/update - Emotional state changes

15. emotion/appraisal/event - Emotional appraisal of events

16. personality/ - Personality simulation related messages

17. personality/state/current - Current personality state
18. personality/expression/communication - Communication style parameters
19. personality/expression/decision - Decision-making parameters

20. personality/expression/emotional - Emotional tendency parameters

21. agency/ - Autonomous agency related messages

22. agency/goals/current - Current agent goals
23. agency/initiative - Proactive engagement initiatives
24. agency/decision/request - Decision-making requests

25. agency/decision/response - Decision outcomes

26. conversation/ - Conversation and dialogue related messages

27. conversation/user/input/v1 - User input messages
28. conversation/ai/response/v1 - AI response messages
29. conversation/context/current - Current conversation context

30. conversation/history/add - Historical conversation data

31. memory/ - Memory and learning related messages

32. memory/store/request - Memory storage requests
33. memory/retrieve/request - Memory retrieval requests
34. memory/consolidation/start - Memory consolidation triggers
35. memory/semantic/query - Semantic memory queries

36. memory/working/update - Working memory updates

37. user/ - User-related messages

38. user/interaction/history - User interaction patterns
39. user/feedback/explicit - Explicit user feedback

40. user/state/update - Inferred user state changes

41. modelservice/ - Model service related messages

42. modelservice/completions/request/v1 - LLM completion requests
43. modelservice/completions/response/v1 - LLM completion responses
44. modelservice/embeddings/request/v1 - Embedding generation requests
45. modelservice/embeddings/response/v1 - Embedding responses
46. modelservice/ner/request/v1 - Named entity recognition requests
47. modelservice/ner/response/v1 - NER responses
48. modelservice/sentiment/request/v1 - Sentiment analysis requests

49. modelservice/sentiment/response/v1 - Sentiment analysis responses

50. ui/ - User Interface related messages

51. ui/state/update - UI state changes (theme, navigation, connection status)
52. ui/interaction/event - User interactions (clicks, input, gestures)
53. ui/notification/show - Display notifications and alerts
54. ui/command/execute - Backend commands to frontend

55. ui/preferences/update - UI preferences and settings updates

56. system/ - System management messages

57. system/bus/started - Message bus startup events
58. system/bus/stopping - Message bus shutdown events
59. system/module/registered - Module registration events

60. system/health - System health checks

61. logs/ - Logging and audit messages

62. logs/backend/main - Backend main process logs
63. logs/backend/api_gateway - API Gateway logs
64. logs/cli/* - All CLI command logs
65. logs/modelservice/* - Modelservice logs
66. logs/* - All log topics (wildcard subscription)

Cross-Cutting Concerns

• crisis/ - Crisis detection and handling
• crisis/detection - Crisis signals and alerts

• crisis/response - Crisis response coordination

• expression/ - Cross-modal expression coordination

• expression/coordination - Coordinated expression directives

• expression/feedback - Expression effectiveness feedback

• learning/ - Shared learning coordination

• learning/coordination - Learning signals and coordination
• learning/feedback - Learning effectiveness feedback

Module Integration Patterns

Publisher-Subscriber Pattern

Modules interact with the message bus through a consistent pattern:

1. Initialization:
2. Modules connect to the message bus on startup
3. They declare topic subscriptions based on their functionality

4. They register message handlers for each subscribed topic

5. Message Publication:

6. Modules publish messages when their internal state changes
7. Messages include standardized metadata and domain-specific payloads

8. Publication is non-blocking and asynchronous

9. Message Consumption:

10. Modules receive messages for their subscribed topics
11. Message handlers process incoming messages
12. Processing may trigger internal state changes or new message publications

Example: Emotion-Personality Integration

The Emotion Simulation and Personality Simulation modules integrate through the message bus:

1. Personality Simulation publishes personality/expression/emotional messages
2. Emotion Simulation subscribes to these messages to adjust emotional tendencies
3. Emotion Simulation publishes emotion/state/current messages
4. Personality Simulation subscribes to these messages to inform personality expression

This bidirectional communication happens without direct dependencies between the modules.

Using the Central Topic Registry

All code should use the AICOTopics class instead of string literals. In practice this looks like:

from aico.core.bus import create_client
from aico.core.topics import AICOTopics

client = create_client("api_gateway")
await client.connect()

await client.publish(AICOTopics.EMOTION_STATE_CURRENT, emotion_data)
await client.subscribe(AICOTopics.CONVERSATION_USER_INPUT, handler)

The TopicMigration helper converts legacy dot-notation topics to the new slash-based scheme for backward compatibility where needed.

Plugin Integration

The Plugin Manager mediates plugin access to the message bus:

1. Topic Access Control:
2. Plugins request access to specific topics
3. Plugin Manager enforces access policies based on plugin permissions

4. Unauthorized topic access attempts are blocked and logged

5. Message Validation:

6. All plugin-originated messages are validated before publication
7. Malformed messages are rejected to prevent system instability

8. Message rate limiting prevents denial-of-service attacks

9. Sandboxed Publication:

10. Plugins publish through the Plugin Manager proxy
11. Messages are tagged with plugin identity for traceability
12. Plugin-specific topic prefixes isolate plugin messages

Security and Privacy Considerations

Message Security

1. CurveZMQ Encryption:
2. Mandatory encryption: All message bus communication uses CurveZMQ with no plaintext fallback
3. Deterministic key derivation: Keys derived from master key using Argon2id + Z85 encoding
4. Mutual authentication: Both broker and clients authenticate using public key cryptography

5. Fail-secure behavior: System fails completely rather than falling back to plaintext

6. Authentication:

7. All modules authenticate to the message bus using CurveZMQ certificates
8. Broker validates specific client public keys (no CURVE_ALLOW_ANY)
9. Unauthorized connections are rejected with comprehensive security logging

10. Plugin authentication uses separate CurveZMQ credentials

11. Authorization:

12. Topic-level access control limits which modules can publish/subscribe
13. Sensitive topics have restricted access
14. Plugin access is limited to approved topics

Privacy Protection

1. Data Minimization:
2. Messages contain only necessary information
3. Sensitive data is filtered before publication

4. User identifiers are anonymized where possible

5. End-to-End Encryption:

6. Transport encryption: All message bus traffic encrypted with CurveZMQ
7. Message payload encryption: Sensitive payloads additionally encrypted at application level
8. Zero plaintext transmission: No unencrypted data crosses network boundaries
9. Key management: Automatic key derivation with secure storage integration

Performance Considerations

Message Throughput

The message bus is designed to handle: - High-frequency emotional state updates - Real-time conversation events - Periodic memory consolidation - Burst traffic during multi-modal coordination

Optimization Strategies

1. Message Prioritization:
2. Critical messages (e.g., crisis detection) receive higher priority

3. Non-time-sensitive messages may be queued during high load

4. Payload Optimization:

5. Large payloads may use compression
6. References instead of full content where appropriate

7. Selective field inclusion for performance-critical paths

8. Subscription Optimization:

9. Fine-grained topic subscriptions to reduce unnecessary message processing
10. Message filtering at the source when possible

11. Local caching of frequently accessed message data

12. Correlation IDs link related messages

13. End-to-end tracing of message flows

14. Timing metrics for message processing

15. Traffic Monitoring:

16. Topic-level message volume metrics
17. Latency measurements for critical paths

18. Queue depth monitoring for backpressure detection

19. Debugging Tools:

20. Message bus inspector for real-time monitoring
21. Message replay capabilities for testing
22. Topic subscription viewer to understand module connectivity

Message Definition and Code Generation

Protocol Buffer Definitions

All message definitions are maintained as Protocol Buffer (.proto) files in the /proto/ directory:

• Core message envelope: /proto/core/envelope.proto
• Emotion messages: /proto/emotion/emotion.proto
• Conversation messages: /proto/conversation/conversation.proto
• Personality messages: /proto/personality/personality.proto
• Integration messages: /proto/integration/integration.proto
• UI messages: /proto/ui/ui.proto

Code Generation Pipeline

The build process automatically generates language-specific code from these definitions:

1. Python classes for backend services
2. Dart classes for Flutter frontend
3. Additional language bindings as needed

CurveZMQ Implementation

AICO's message bus implements mandatory CurveZMQ encryption for all inter-component communication with the following core principles:

1. Mandatory Encryption: No plaintext fallback – system fails securely if encryption cannot be established.
2. Mutual Authentication: Broker and clients authenticate using public key cryptography.
3. Deterministic Key Derivation: All keys derived from the master key using Argon2id + Z85 encoding.
4. Fail-Secure Design: Encryption failures result in system failure, not insecure fallback.

Key management, broker/client configuration, socket options, and security logging are all encapsulated in MessageBusBroker, MessageBusClient, and AICOKeyManager, so most code only needs to create a client and connect.

Testing and Validation

Encryption and message bus behavior can be verified via the existing test script (scripts/test_curve_zmq.py) and the aico bus CLI commands (test, monitor, stats). Detailed command output is omitted here to keep the architecture doc focused.

Migration from Plaintext

Removed Components

1. Plaintext fallback code: All fallback mechanisms removed
2. CURVE_ALLOW_ANY: Replaced with explicit client authentication
3. Raw ZMQ sockets: All components use encrypted MessageBusClient
4. IPC adapter: Unused ZeroMQ IPC adapter removed

Breaking Changes

• No backward compatibility: Old plaintext clients cannot connect
• Master key required: All components require master key for operation
• Fail-secure only: No graceful degradation to plaintext mode

Troubleshooting

Common Issues

Authentication Failures:

[SECURITY] CRITICAL: Failed to setup CurveZMQ authentication

Solution: Verify master key is available and AICOKeyManager is properly configured.

Key Derivation Errors:

[SECURITY] CRITICAL: Failed to setup CurveZMQ encryption

Solution: Check master key authentication and key manager initialization.

Connection Refused:

MessageBusError: CurveZMQ socket configuration failed

Solution: Ensure broker is running and client public key is in authorized list.

Debug Logging

Debug logging for the message bus can be enabled via the standard Python logging configuration on the aico.core.bus logger.

Security Guarantees

What is Protected

✅ All message bus traffic encrypted
✅ Mutual authentication between all components
✅ No plaintext fallback possible
✅ Deterministic key derivation from master key
✅ Comprehensive security logging

What is NOT Protected

❌ Application-level message content (use additional encryption if needed)
❌ Topic names (visible in ZeroMQ subscription filters)
❌ Message timing/frequency (traffic analysis still possible)

Performance Impact

Encryption Overhead

• CPU: ~5-10% overhead for CurveZMQ encryption/decryption
• Memory: Minimal additional memory usage
• Latency: <1ms additional latency per message
• Throughput: >95% of plaintext performance maintained

Optimization Tips

1. Reuse connections: Avoid frequent connect/disconnect cycles
2. Batch messages: Group small messages when possible
3. Monitor key derivation: Cache derived keys when appropriate

Conclusion

The Core Message Bus architecture is fundamental to AICO's modular, event-driven design. It enables:

• Modularity: Components can be developed, tested, and deployed independently
• Extensibility: New modules and plugins can be integrated without modifying existing code
• Resilience: Failures in one module don't cascade to others
• Adaptability: The system can evolve through versioned message formats
• Autonomy: Modules can operate independently based on events
• Performance: Binary serialization optimizes for speed and size
• Cross-Platform: Consistent message format across all platforms and devices
• Security: Mandatory CurveZMQ encryption ensures all communication is protected

By providing a standardized, secure communication backbone, the message bus facilitates the complex interactions required for AICO's proactive agency, emotional presence, personality consistency, and multi-modal embodiment across its federated device network.