16. Notification System

A personal AI agent must communicate with users through a notification system that balances awareness with intrusion avoidance. OpenCLaw implements multi-channel notifications with intelligent batching and priority classification.

Priority Classification

Notifications classify into four priority levels: critical (requires immediate attention), elevated (important but not urgent), normal (informational), and low (non-essential updates).

# notification_system.py
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional

class NotificationPriority(Enum):
    CRITICAL = 1
    ELEVATED = 2
    NORMAL = 3
    LOW = 4

@dataclass
class Notification:
    id: str
    title: str
    body: str
    priority: NotificationPriority
    channel: str
    timestamp: datetime
    metadata: dict
    acknowledged: bool = False
    
class PriorityClassifier:
    def classify(self, event: Event) -> NotificationPriority:
        if event.is_critical:
            return NotificationPriority.CRITICAL
        
        if event.user_availability == Availability.AWAY:
            return NotificationPriority.NORMAL
        
        if event.urgency_score > 0.8:
            return NotificationPriority.ELEVATED
        
        return NotificationPriority.NORMAL

Channel Management

OpenCLaw supports multiple notification channels: push notifications (mobile), email (synchronous external), in-app messages (synchronous internal), and webhook callbacks (machine-to-machine).

class NotificationChannel:
    def send(self, notification: Notification) -> bool:
        raise NotImplementedError

class PushChannel(NotificationChannel):
    def __init__(self, push_service):
        self.service = push_service
    
    def send(self, notification: Notification) -> bool:
        if notification.priority.value > NotificationPriority.NORMAL.value:
            return self.service.send_immediate(notification)
        return self.service.queue(notification)

class EmailChannel(NotificationChannel):
    def __init__(self, email_client):
        self.client = email_client
    
    def send(self, notification: Notification) -> bool:
        if notification.priority <= NotificationPriority.ELEVATED:
            return self.client.send_immediate(
                notification.title,
                notification.body
            )
        return self.client.queue_digest(
            [notification],
            delay_hours=4
        )

class ChannelRouter:
    def __init__(self, channels: List[NotificationChannel]):
        self.channels = {c.name: c for c in channels}
    
    def route(self, notification: Notification) -> bool:
        channel = self.channels.get(notification.channel)
        if not channel:
            channel = self.select_default_channel(notification)
        
        return channel.send(notification)

Batching and Digest

Low and normal priority notifications batch into digests to reduce interruption frequency. Digest timing respects user-configured quiet hours and activity patterns.

class NotificationBatcher:
    def __init__(
        self,
        batch_window: timedelta = timedelta(hours=1),
        max_batch_size: int = 10
    ):
        self.batch_window = batch_window
        self.max_batch_size = max_batch_size
        self.pending = []
        self.last_flush = datetime.utcnow()
    
    def add(self, notification: Notification) -> None:
        if notification.priority <= NotificationPriority.ELEVATED:
            self.flush_immediate(notification)
            return
        
        self.pending.append(notification)
        
        if len(self.pending) >= self.max_batch_size:
            self.flush()
    
    def flush(self) -> List[Notification]:
        batch = self.pending
        self.pending = []
        self.last_flush = datetime.utcnow()
        return batch
    
    def should_auto_flush(self) -> bool:
        elapsed = datetime.utcnow() - self.last_flush
        return elapsed > self.batch_window