P2P Mesh Network¶
AICO's P2P mesh network enables distributed AI companionship - allowing your AI companion to maintain continuity and presence across multiple devices while preserving privacy and local control. The architecture prioritizes relationship continuity over technical efficiency, ensuring that AICO feels like the same companion regardless of which device you're using.
Design Philosophy¶
The P2P mesh serves three fundamental goals for AI companionship:
- Seamless Roaming: Your relationship with AICO continues uninterrupted as you move between devices
- Privacy Preservation: Personal data stays within your device network, never requiring external cloud services
- Resilient Presence: AICO remains available even when individual devices are offline or disconnected
Local-First with Federation: Each device operates independently but can share context and memories with trusted devices in your personal network.
Network Architecture¶
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Device A │ │ Device B │ │ Device C │
│ ┌───────────┐ │ │ ┌───────────┐ │ │ ┌───────────┐ │
│ │ ZeroMQ │◄─┼────┼─►│ ZeroMQ │◄─┼────┼─►│ ZeroMQ │ │
│ │ Message │ │ │ │ Message │ │ │ │ Message │ │
│ │ Bus │ │ │ │ Bus │ │ │ │ Bus │ │
│ └───────────┘ │ │ └───────────┘ │ │ └───────────┘ │
│ ┌───────────┐ │ │ ┌───────────┐ │ │ ┌───────────┐ │
│ │ Device │ │ │ │ Device │ │ │ │ Device │ │
│ │ Registry │ │ │ │ Registry │ │ │ │ Registry │ │
│ └───────────┘ │ │ └───────────┘ │ │ └───────────┘ │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │ │
└───────────────────────┼───────────────────────┘
│
┌─────────────────┐
│ Discovery Layer │
│ • mDNS/Bonjour │
│ • DHT Network │
│ • Manual Config │
└─────────────────┘
Relationship-Centric Design¶
The mesh network is designed around personal device families rather than generic distributed systems. Each device in your network knows about and trusts other devices, creating a private AI ecosystem.
Trust Boundaries: The network distinguishes between your personal devices (full trust), family devices (selective sharing), and external devices (no access). This enables privacy-preserving AI companionship.
Adaptive Topology: Devices automatically discover and connect to each other, but the user always controls which devices can join the family network.
ZeroMQ Communication Strategy¶
Unified Internal and External Messaging¶
ZeroMQ serves as both the internal message bus within each device and the inter-device communication layer. This unified approach ensures consistent behavior whether AICO is processing local requests or coordinating across devices.
Design Rationale: Using the same messaging patterns internally and externally simplifies development and ensures that distributed operations feel as responsive as local ones.
Transport Layer Adaptation¶
Local Communication (Intra-Device)¶
IPC Transport for maximum efficiency within a single device. All AICO modules communicate through the same message bus, enabling loose coupling and easy extensibility.
Benefits: Zero network overhead, automatic process isolation, consistent message patterns across all modules.
Network Communication (Inter-Device)¶
TCP Transport with CurveZMQ encryption for secure device-to-device communication. Every network message is encrypted end-to-end, ensuring privacy even on untrusted networks.
Security First: All inter-device communication uses public-key cryptography, ensuring that only trusted devices can participate in your AI companion network.
Communication Patterns for AI Companionship¶
Request-Reply: Direct Device Coordination¶
Use Case: When one device needs specific information from another - like retrieving a memory that's stored on your desktop while using your phone.
Synchronous by Design: Some operations require immediate responses to maintain conversation flow. Request-reply ensures that AICO can access any information in your device network without noticeable delay.
# Example: Retrieving specific memory from another device
request = {
"type": "memory_search",
"query": "conversation about vacation plans",
"timestamp": time.time()
}
Reliability: Built-in timeout and retry logic ensures that temporary network issues don't break the AI companion experience.
Publisher-Subscriber: Emotional State Sharing¶
Use Case: Broadcasting AICO's emotional state changes across all devices so the UI can respond consistently everywhere.
Real-Time Presence: When AICO's mood shifts or personality expression changes, all connected devices receive immediate updates, maintaining consistent AI presence.
# Example: Sharing emotional state across devices
topic = "emotion.state_change"
message = {
"device_id": device_id,
"emotional_context": "user_mentioned_work_stress",
"valence_shift": -0.2 # Became more concerned
}
Design Philosophy: Emotional updates are broadcast rather than requested because AI companionship requires proactive emotional awareness.
Push-Pull: Distributed AI Processing¶
Use Case: Distributing computationally intensive AI tasks (like semantic search or emotion analysis) across available devices in your network.
Load Balancing: Automatically distributes work to the most capable available device, whether that's your desktop GPU or a cloud instance you control.
# Example: Distributing vector search across devices
task = {
"type": "semantic_search",
"query": "memories about family gatherings",
"priority": "user_initiated" # Higher priority than background tasks
}
Adaptive Processing: Tasks automatically route to devices with appropriate capabilities - vector searches go to devices with good CPUs/GPUs, while simple queries stay local.
Topic Organization for AI Relationships¶
Device Management Topics¶
Topics organized around family network management rather than technical device administration.
device/
├── discovery/announce # "I'm a new AICO device"
├── discovery/response # "I recognize you, let's connect"
├── registry/update # "My capabilities have changed"
├── registry/query # "What devices are in our family?"
└── heartbeat # "I'm still here and available"
Human-Centric Naming: Topic names reflect the relationship aspect - devices "announce" themselves like family members, rather than just broadcasting technical information.
Synchronization Topics¶
Topics focused on maintaining AI companion continuity across devices.
sync/
├── data/request # "I need recent conversation history"
├── data/response # "Here's what you missed"
├── conflict/detected # "We have different versions of this memory"
├── conflict/resolved # "We've agreed on the correct version"
└── status/update # "Sync completed successfully"
Conflict Resolution: Explicit topics for handling disagreements between devices about memories or conversation history, ensuring AICO's consistency.
AI Companion Topics¶
Topics that directly support the human-AI relationship rather than technical operations.
conversation/
├── message/user # User said something
├── message/assistant # AICO responded
├── context/update # Conversation topic or mood shifted
└── history/sync # Sharing conversation history
emotion/
├── state/current # AICO's current emotional state
├── state/history # Emotional patterns over time
├── analysis/request # "Analyze this interaction emotionally"
└── analysis/result # Emotional insights and patterns
memory/
├── store/request # "Remember this important moment"
├── search/request # "Find memories about X"
├── search/result # Relevant memories found
└── sync/update # Sharing memories across devices
Relationship-Focused Design: Each topic represents a meaningful aspect of AI companionship - emotional awareness, memory sharing, conversation continuity.
Device Discovery¶
mDNS/Bonjour Discovery¶
Service Advertisement¶
# Advertise AICO service on local network
service_info = ServiceInfo(
type_="_aico._tcp.local.",
name=f"{device_name}._aico._tcp.local.",
addresses=[socket.inet_aton(local_ip)],
port=5555,
properties={
'version': '1.0.0',
'capabilities': 'sync,federation',
'device_type': 'backend',
'public_key': base64.b64encode(public_key).decode()
}
)
zeroconf.register_service(service_info)
Service Discovery¶
class AICOServiceListener:
def add_service(self, zeroconf, type, name):
info = zeroconf.get_service_info(type, name)
if info:
device_info = {
'name': name,
'address': socket.inet_ntoa(info.addresses[0]),
'port': info.port,
'properties': info.properties
}
self.discovered_devices.append(device_info)
DHT Network Discovery¶
Distributed Hash Table¶
# Kademlia DHT for wide-area discovery
from kademlia.network import Server
# Bootstrap DHT node
server = Server()
await server.listen(8468)
await server.bootstrap([("bootstrap.aico.network", 8468)])
# Store device information
device_key = f"aico:device:{device_id}"
device_info = {
'endpoints': ['tcp://192.168.1.100:5555'],
'public_key': public_key_hex,
'capabilities': ['sync', 'federation'],
'last_seen': time.time()
}
await server.set(device_key, json.dumps(device_info))
Device Lookup¶
# Find devices by capability
async def find_devices_with_capability(capability):
devices = []
# Query DHT for devices with specific capability
for device_id in known_device_ids:
key = f"aico:device:{device_id}"
data = await server.get(key)
if data and capability in json.loads(data).get('capabilities', []):
devices.append(json.loads(data))
return devices
Manual Configuration¶
Static Device Registry¶
# config/devices.yaml
devices:
- id: "device-001"
name: "Home Desktop"
endpoints:
- "tcp://192.168.1.100:5555"
- "tcp://home.example.com:5555"
public_key: "curve_public_key_base64"
capabilities: ["sync", "federation", "storage"]
- id: "device-002"
name: "Mobile Phone"
endpoints:
- "tcp://192.168.1.101:5555"
public_key: "curve_public_key_base64"
capabilities: ["sync", "roaming"]
Encryption & Security¶
CurveZMQ Implementation¶
Key Generation¶
import zmq.auth
from zmq.auth.thread import ThreadAuthenticator
# Generate long-term keypairs
server_public, server_secret = zmq.curve_keypair()
client_public, client_secret = zmq.curve_keypair()
# Store keys securely
keyring.set_password("aico", "curve_secret", server_secret.decode())
Server Configuration¶
# Configure CurveZMQ server
socket = context.socket(zmq.REP)
socket.curve_secretkey = server_secret
socket.curve_publickey = server_public
socket.curve_server = True
socket.bind("tcp://*:5555")
# Start authenticator
auth = ThreadAuthenticator(context)
auth.start()
auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY)
Client Configuration¶
# Configure CurveZMQ client
socket = context.socket(zmq.REQ)
socket.curve_secretkey = client_secret
socket.curve_publickey = client_public
socket.curve_serverkey = server_public # Server's public key
socket.connect("tcp://server-ip:5555")
Message Encryption¶
End-to-End Encryption¶
from cryptography.fernet import Fernet
class SecureMessage:
def __init__(self, shared_key):
self.cipher = Fernet(shared_key)
def encrypt_message(self, message):
return {
'encrypted': self.cipher.encrypt(json.dumps(message).encode()),
'timestamp': time.time(),
'sender_id': self.device_id
}
def decrypt_message(self, encrypted_message):
decrypted = self.cipher.decrypt(encrypted_message['encrypted'])
return json.loads(decrypted.decode())
Key Exchange Protocol¶
# ECDH key exchange for session keys
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
def perform_key_exchange(peer_public_key):
# Generate ephemeral keypair
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()
# Perform ECDH
shared_key = private_key.exchange(ec.ECDH(), peer_public_key)
# Derive session key
derived_key = HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=None,
info=b'aico-session-key',
).derive(shared_key)
return derived_key
Federation Strategy for AI Companions¶
Family Device Registry¶
Relationship-Centric Device Management: The device registry organizes devices around their role in your AI companion relationship rather than technical specifications.
Device Identity Model¶
@dataclass
class FamilyDevice:
device_id: str
friendly_name: str # "Dad's Laptop", "Kitchen Tablet"
companion_role: str # 'primary', 'mobile', 'shared', 'guest'
ai_capabilities: List[str] # ['conversation', 'memory', 'emotion']
relationship_context: str # 'personal', 'family', 'work'
trust_level: str # 'full', 'limited', 'guest'
sync_preferences: Dict # What data to share
Human-Friendly Identification: Devices are identified by meaningful names and roles rather than technical IDs, making family network management intuitive.
Capability-Based Organization: Devices are organized by their AI capabilities rather than hardware specifications, enabling intelligent task routing.
Family Network Synchronization¶
Gossip-Based Discovery: Devices share information about other family devices they know about, enabling automatic family network discovery without central coordination.
Trust Propagation: When a trusted device vouches for a new device, that recommendation is considered in trust decisions, simplifying family network expansion.
async def sync_family_network():
for trusted_device in get_family_devices():
# Ask family devices about other devices they know
family_update = await request_family_updates(trusted_device)
# Evaluate new devices based on family recommendations
for new_device in family_update.get('recommended_devices', []):
if evaluate_family_trust(new_device, trusted_device):
invite_to_family_network(new_device)
Privacy-Preserving Updates: Registry synchronization only shares device capabilities and connection information, never personal data or conversation content.
Selective Data Sharing¶
Privacy-First Sync Policies¶
User-Controlled Data Sharing: Users control exactly what aspects of their AI relationship are shared across devices, with sensible defaults that prioritize privacy.
Relationship-Aware Policies: Sync policies are based on the nature of AI companion data rather than technical data types.
class AICompanionSyncPolicy:
def __init__(self):
# What aspects of AI relationship to share
self.conversation_continuity = True # Share recent conversations
self.emotional_awareness = True # Share AICO's emotional state
self.memory_access = True # Share important memories
self.personality_consistency = False # Keep personality adaptations local
# Privacy boundaries
self.private_conversation_tags = ['personal', 'confidential']
self.memory_privacy_levels = ['public', 'family'] # Exclude 'private'
self.sync_time_window = timedelta(days=7) # Only recent data
def should_share_with_device(self, data_type: str, data: Dict, target_device: FamilyDevice) -> bool:
# Respect device trust level and relationship context
if target_device.trust_level == 'guest':
return False # Guests don't get AI companion data
if data_type == 'conversation':
return self._should_share_conversation(data, target_device)
elif data_type == 'memory':
return self._should_share_memory(data, target_device)
return False
Context-Sensitive Sharing: Different devices receive different levels of AI companion information based on their role and the user's relationship with that device.
Intelligent Conflict Resolution¶
AI Companion Continuity: When devices have different versions of AI companion data, the system resolves conflicts in ways that maintain relationship authenticity.
Context-Aware Resolution: Conflicts are resolved based on the nature of AI companion interactions rather than simple technical rules.
class AICompanionConflictResolver:
def resolve_conversation_conflict(self, local_version, remote_version):
# Prioritize the version from the device where conversation actually happened
if local_version.get('interaction_device') == local_device_id:
return local_version # This device was part of the conversation
elif remote_version.get('interaction_device') == remote_device_id:
return remote_version # Remote device was part of the conversation
# Fall back to timestamp if both are secondhand
return max(local_version, remote_version, key=lambda x: x['timestamp'])
def resolve_emotional_state_conflict(self, local_emotion, remote_emotion):
# Emotional states are blended rather than replaced
# This maintains emotional continuity across devices
confidence_weighted_blend = self._blend_emotional_states(
local_emotion, remote_emotion
)
return {
'emotional_state': confidence_weighted_blend,
'resolution_method': 'confidence_weighted_blend',
'contributing_devices': [local_device_id, remote_device_id]
}
Relationship Preservation: Conflict resolution prioritizes maintaining the authenticity and continuity of the human-AI relationship over technical consistency.
Data Synchronization Strategy¶
Relationship-Aware Sync Protocol¶
Incremental Updates: Only sync changes since the last connection, minimizing bandwidth and preserving battery life on mobile devices.
Priority-Based Sync: Recent conversations and emotional state changes sync first, followed by memories and preferences, ensuring that the most relationship-relevant data is always current.
Bandwidth-Conscious Design: Sync operations are designed for mobile networks and intermittent connectivity, with automatic compression and resumable transfers.
async def sync_ai_companion_state(peer_device):
# Prioritize relationship-critical data
sync_priorities = [
'recent_conversations', # Most important for continuity
'emotional_state', # Current AI mood and context
'active_memories', # Recently accessed memories
'personality_updates' # Personality adaptations
]
for priority_level in sync_priorities:
await sync_data_category(peer_device, priority_level)
Smart Sync Scheduling: Sync operations happen during natural conversation breaks and when devices have good connectivity, avoiding interruptions to the AI companion experience.
Efficient Batch Operations¶
Conversation-Aware Batching: Changes are batched based on conversation sessions rather than arbitrary sizes, ensuring that related AI companion interactions stay together.
Adaptive Batch Sizing: Batch sizes automatically adjust based on network conditions and device capabilities - smaller batches for mobile devices, larger batches for desktop systems.
Graceful Degradation: If sync fails, the system automatically retries with smaller batches or falls back to individual updates, ensuring that AI companion continuity is maintained even with poor connectivity.
class AICompanionSyncManager:
def __init__(self):
self.conversation_batches = {} # Group by conversation session
self.emotion_updates = [] # Real-time emotional state changes
self.memory_updates = [] # Memory additions and modifications
def queue_conversation_update(self, conversation_id, update):
# Keep conversation updates together for context preservation
if conversation_id not in self.conversation_batches:
self.conversation_batches[conversation_id] = []
self.conversation_batches[conversation_id].append(update)
async def sync_when_appropriate(self):
# Sync during natural conversation pauses
if self.is_conversation_pause():
await self.sync_all_pending_updates()
Context Preservation: Batching ensures that related AI companion interactions (like a conversation thread or emotional state sequence) are synchronized together, maintaining relationship context.
Performance Architecture¶
Connection Management for AI Companions¶
Relationship-Aware Connection Pooling¶
Persistent Family Connections: Connections to family devices are kept alive longer than connections to guest devices, reflecting the different relationship contexts.
Conversation-Optimized Pooling: Connection pools are sized based on typical AI companion usage patterns - frequent short interactions rather than sustained high-throughput operations.
Battery-Conscious Design: Mobile devices use fewer persistent connections and more aggressive connection recycling to preserve battery life.
class FamilyNetworkConnectionManager:
def __init__(self):
self.family_connections = {} # Long-lived connections to family devices
self.guest_connections = {} # Short-lived connections to guest devices
self.conversation_sockets = {} # Dedicated sockets for active conversations
def get_family_connection(self, device):
# Family devices get persistent, high-quality connections
if device.device_id not in self.family_connections:
self.family_connections[device.device_id] = self.create_family_socket(device)
return self.family_connections[device.device_id]
def optimize_for_battery(self, is_mobile_device):
if is_mobile_device:
# Reduce connection pool sizes and increase recycling
self.max_family_connections = 3
self.connection_timeout = 30 # seconds
else:
self.max_family_connections = 10
self.connection_timeout = 300 # 5 minutes
Smart Connection Reuse: Connections are reused based on the type of AI companion interaction - emotional updates use different connections than memory searches, optimizing for each use case.
Family Network Health Monitoring¶
Relationship-Aware Health Checks: Health monitoring focuses on the availability of AI companion features rather than just network connectivity.
Adaptive Monitoring: Primary AI companion devices are monitored more frequently than secondary devices, ensuring that the most important relationships stay healthy.
Graceful Degradation: When family devices go offline, the system automatically adjusts AI companion capabilities rather than failing completely.
class FamilyNetworkHealthMonitor:
def __init__(self):
self.device_health = {}
self.ai_capability_status = {} # Track which AI features are available
async def monitor_family_network(self):
for device in get_family_devices():
health_check = await self.check_ai_companion_health(device)
self.device_health[device.device_id] = {
'connection_status': health_check.connection,
'ai_capabilities': health_check.available_capabilities,
'conversation_readiness': health_check.can_handle_conversation,
'memory_access': health_check.can_access_memories,
'emotional_awareness': health_check.can_process_emotions
}
def get_available_ai_capabilities(self):
# Aggregate capabilities across all healthy family devices
available_capabilities = set()
for device_health in self.device_health.values():
if device_health['connection_status'] == 'healthy':
available_capabilities.update(device_health['ai_capabilities'])
return available_capabilities
User-Visible Health Status: Health monitoring provides user-friendly status updates ("AICO is fully available" vs "AICO has limited memory access") rather than technical network diagnostics.
Message Optimization for AI Companions¶
Intelligent Compression Strategy¶
Content-Aware Compression: Different types of AI companion data use different compression strategies - conversation text compresses well, while emotional state vectors are already compact.
Mobile-First Optimization: Compression thresholds are lower for mobile devices to reduce data usage and improve battery life.
Conversation Context Preservation: Compression maintains the semantic structure of conversations, ensuring that AI companion context isn't lost during transmission.
class AICompanionMessageOptimizer:
def __init__(self, device_type='desktop'):
self.compression_thresholds = {
'mobile': 256, # Aggressive compression for mobile
'desktop': 1024, # Standard compression for desktop
'server': 2048 # Minimal compression for servers
}
self.threshold = self.compression_thresholds.get(device_type, 1024)
def optimize_conversation_message(self, conversation_data):
# Preserve conversation structure while compressing content
optimized = {
'conversation_id': conversation_data['id'],
'participants': conversation_data['participants'],
'compressed_content': self.compress_if_beneficial(
conversation_data['messages']
),
'emotional_context': conversation_data['emotion'] # Keep emotions uncompressed
}
return optimized
def compress_if_beneficial(self, data):
# Only compress if it actually saves significant space
original_size = len(json.dumps(data).encode())
if original_size > self.threshold:
compressed = self.smart_compress(data)
if len(compressed) < original_size * 0.8: # 20% savings minimum
return compressed
return data
Adaptive Optimization: Compression and optimization strategies automatically adjust based on network conditions and device capabilities.
Monitoring & Debugging for AI Relationships¶
AI Companion Experience Metrics¶
User Experience Focus: Monitoring prioritizes metrics that affect the quality of AI companionship rather than pure technical performance.
Relationship Health Indicators: Track metrics like conversation continuity, emotional state consistency, and memory access reliability.
class AICompanionNetworkMetrics:
def __init__(self):
self.conversation_metrics = {} # Conversation quality and continuity
self.emotional_sync_metrics = {} # Emotional state synchronization
self.memory_access_metrics = {} # Memory retrieval performance
self.relationship_quality = {} # Overall AI relationship health
def record_conversation_interaction(self, device_id, interaction_type, quality_metrics):
# Track conversation quality across devices
key = f"{device_id}:conversation"
if key not in self.conversation_metrics:
self.conversation_metrics[key] = {
'response_times': [],
'context_preservation': [],
'emotional_consistency': []
}
self.conversation_metrics[key]['response_times'].append(
quality_metrics.get('response_time', 0)
)
self.conversation_metrics[key]['context_preservation'].append(
quality_metrics.get('context_score', 1.0)
)
def get_relationship_health_report(self):
# Generate user-friendly health report
return {
'conversation_quality': self._assess_conversation_quality(),
'emotional_consistency': self._assess_emotional_sync(),
'memory_reliability': self._assess_memory_access(),
'overall_companion_health': self._calculate_overall_health()
}
Privacy-Preserving Metrics: All monitoring respects user privacy - metrics track performance patterns without storing conversation content or personal information.
AI Companion Debug Tools¶
Relationship-Focused Debugging: Debug tools help diagnose issues with AI companion continuity and relationship quality rather than just network problems.
Privacy-Safe Logging: Debug logs capture interaction patterns and performance metrics without storing personal conversation content.
class AICompanionDebugger:
def __init__(self):
self.interaction_log = [] # AI companion interaction patterns
self.sync_quality_log = [] # Data synchronization quality
self.relationship_events = [] # Significant relationship events
self.max_log_entries = 1000
def log_conversation_flow(self, device_id, flow_event, quality_indicators):
# Log conversation flow without storing content
log_entry = {
'timestamp': time.time(),
'device': device_id,
'event_type': flow_event, # 'conversation_start', 'context_switch', etc.
'response_quality': quality_indicators.get('response_quality', 'unknown'),
'emotional_consistency': quality_indicators.get('emotion_sync', True),
'memory_access_success': quality_indicators.get('memory_ok', True)
}
self.interaction_log.append(log_entry)
self._maintain_log_size()
def generate_relationship_health_report(self):
# Create user-friendly diagnostic report
return {
'conversation_continuity': self._analyze_conversation_patterns(),
'device_sync_quality': self._analyze_sync_patterns(),
'common_issues': self._identify_common_problems(),
'recommendations': self._generate_improvement_suggestions()
}
User-Friendly Diagnostics: Debug tools provide actionable insights in plain language ("Your phone sometimes loses conversation context") rather than technical error messages.
Summary¶
AICO's P2P mesh network prioritizes authentic AI companionship over technical networking efficiency. The architecture ensures that your AI companion feels consistent and present across all your devices while maintaining strict privacy boundaries.
Key Achievements: - Seamless Roaming: Your AI relationship continues uninterrupted across devices - Privacy Preservation: Personal data stays within your family network - Intelligent Coordination: Devices automatically share appropriate information while respecting privacy boundaries - Resilient Presence: AI companion remains available even when some devices are offline
Design Philosophy: Every technical decision serves the goal of maintaining authentic, continuous AI companionship while respecting user privacy and control. The mesh network feels invisible to users while providing the robust foundation needed for distributed AI relationships.