""" AIVA Prospective Memory System ============================== A production-grade prospective memory system that enables AIVA to remember and execute future intentions. Implements cognitive science principles for event-based and time-based prospective memory with robust monitoring. Components: - IntentionStore: Persistent storage for future intentions - EventBasedCue: Triggers intentions based on contextual events - TimeBasedCue: Triggers intentions at specific times - MonitoringProcess: Background monitoring for cue detection - RetrievalProcess: Intelligent intention retrieval - ExecutionTracker: Track and verify intention execution Author: Genesis Lead Architect Version: 1.0.0 """ import asyncio import json import logging import threading import time import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from enum import Enum, auto from pathlib import Path from typing import ( Any, Callable, Dict, List, Optional, Set, Tuple, Union, TypeVar, Generic ) from collections import defaultdict from functools import wraps import hashlib import re # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("ProspectiveMemory") # ============================================================================= # ENUMS AND CONSTANTS # ============================================================================= class IntentionStatus(Enum): """Status of an intention in the prospective memory system.""" PENDING = "pending" ACTIVE = "active" TRIGGERED = "triggered" EXECUTED = "executed" FAILED = "failed" EXPIRED = "expired" CANCELLED = "cancelled" class CueType(Enum): """Types of prospective memory cues.""" EVENT_BASED = auto() # Triggered by external events TIME_BASED = auto() # Triggered at specific times LOCATION_BASED = auto() # Triggered by location context ACTIVITY_BASED = auto() # Triggered by activity patterns HYBRID = auto() # Combination of multiple cue types class Priority(Enum): """Priority levels for intentions.""" CRITICAL = 1 HIGH = 2 MEDIUM = 3 LOW = 4 BACKGROUND = 5 class ExecutionResult(Enum): """Result of intention execution.""" SUCCESS = "success" PARTIAL = "partial" FAILED = "failed" DEFERRED = "deferred" SKIPPED = "skipped" # ============================================================================= # DATA CLASSES # ============================================================================= @dataclass class Intention: """ Represents a future intention to be remembered and executed. Attributes: id: Unique identifier for the intention content: Description of what needs to be done action: The action to execute (callable name or function) action_params: Parameters for the action cue_type: Type of cue that triggers this intention cue_conditions: Conditions that must be met for triggering created_at: When the intention was created trigger_time: For time-based cues, when to trigger expiry_time: When the intention expires priority: Priority level status: Current status metadata: Additional metadata execution_count: Number of execution attempts last_execution: Last execution timestamp parent_id: Parent intention ID for hierarchical intentions tags: Tags for categorization """ id: str = field(default_factory=lambda: str(uuid.uuid4())) content: str = "" action: str = "" action_params: Dict[str, Any] = field(default_factory=dict) cue_type: CueType = CueType.EVENT_BASED cue_conditions: Dict[str, Any] = field(default_factory=dict) created_at: float = field(default_factory=time.time) trigger_time: Optional[float] = None expiry_time: Optional[float] = None priority: Priority = Priority.MEDIUM status: IntentionStatus = IntentionStatus.PENDING metadata: Dict[str, Any] = field(default_factory=dict) execution_count: int = 0 last_execution: Optional[float] = None parent_id: Optional[str] = None tags: Set[str] = field(default_factory=set) def to_dict(self) -> Dict[str, Any]: """Convert intention to dictionary for serialization.""" return { "id": self.id, "content": self.content, "action": self.action, "action_params": self.action_params, "cue_type": self.cue_type.name, "cue_conditions": self.cue_conditions, "created_at": self.created_at, "trigger_time": self.trigger_time, "expiry_time": self.expiry_time, "priority": self.priority.name, "status": self.status.value, "metadata": self.metadata, "execution_count": self.execution_count, "last_execution": self.last_execution, "parent_id": self.parent_id, "tags": list(self.tags) } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Intention": """Create intention from dictionary.""" return cls( id=data.get("id", str(uuid.uuid4())), content=data.get("content", ""), action=data.get("action", ""), action_params=data.get("action_params", {}), cue_type=CueType[data.get("cue_type", "EVENT_BASED")], cue_conditions=data.get("cue_conditions", {}), created_at=data.get("created_at", time.time()), trigger_time=data.get("trigger_time"), expiry_time=data.get("expiry_time"), priority=Priority[data.get("priority", "MEDIUM")], status=IntentionStatus(data.get("status", "pending")), metadata=data.get("metadata", {}), execution_count=data.get("execution_count", 0), last_execution=data.get("last_execution"), parent_id=data.get("parent_id"), tags=set(data.get("tags", [])) ) def is_expired(self) -> bool: """Check if intention has expired.""" if self.expiry_time is None: return False return time.time() > self.expiry_time def is_ready_for_time_trigger(self) -> bool: """Check if time-based intention is ready to trigger.""" if self.cue_type != CueType.TIME_BASED: return False if self.trigger_time is None: return False return time.time() >= self.trigger_time @dataclass class ExecutionRecord: """ Record of an intention execution attempt. Attributes: id: Unique execution record ID intention_id: ID of the intention being executed timestamp: When execution occurred result: Execution result duration_ms: Execution duration in milliseconds error: Error message if failed output: Output from the action context: Context at time of execution """ id: str = field(default_factory=lambda: str(uuid.uuid4())) intention_id: str = "" timestamp: float = field(default_factory=time.time) result: ExecutionResult = ExecutionResult.SUCCESS duration_ms: float = 0.0 error: Optional[str] = None output: Any = None context: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary.""" return { "id": self.id, "intention_id": self.intention_id, "timestamp": self.timestamp, "result": self.result.value, "duration_ms": self.duration_ms, "error": self.error, "output": str(self.output) if self.output else None, "context": self.context } @dataclass class CueMatch: """Represents a match between an event and a cue condition.""" intention_id: str match_score: float # 0.0 to 1.0 matched_conditions: Dict[str, Any] timestamp: float = field(default_factory=time.time) # ============================================================================= # INTENTION STORE # ============================================================================= class IntentionStore: """ Persistent storage for future intentions. Features: - File-based persistence with JSON serialization - In-memory indexing for fast retrieval - Automatic expiry management - Tag-based organization - Priority queuing """ def __init__( self, storage_path: Optional[str] = None, auto_persist: bool = True, persist_interval: float = 30.0 ): """ Initialize the intention store. Args: storage_path: Path to persistence file auto_persist: Enable automatic persistence persist_interval: Seconds between auto-persist """ self.storage_path = Path(storage_path) if storage_path else None self.auto_persist = auto_persist self.persist_interval = persist_interval # Main storage self._intentions: Dict[str, Intention] = {} # Indexes for fast lookup self._by_status: Dict[IntentionStatus, Set[str]] = defaultdict(set) self._by_priority: Dict[Priority, Set[str]] = defaultdict(set) self._by_cue_type: Dict[CueType, Set[str]] = defaultdict(set) self._by_tag: Dict[str, Set[str]] = defaultdict(set) self._time_sorted: List[str] = [] # Sorted by trigger_time # Threading self._lock = threading.RLock() self._persist_thread: Optional[threading.Thread] = None self._running = False # Stats self._stats = { "total_created": 0, "total_executed": 0, "total_failed": 0, "total_expired": 0 } # Load existing data if self.storage_path and self.storage_path.exists(): self._load_from_disk() # Start auto-persist if self.auto_persist: self._start_persist_thread() def store(self, intention: Intention) -> str: """ Store a new intention. Args: intention: The intention to store Returns: The intention ID """ with self._lock: self._intentions[intention.id] = intention self._update_indexes(intention) self._stats["total_created"] += 1 logger.info(f"Stored intention: {intention.id} - {intention.content[:50]}") return intention.id def get(self, intention_id: str) -> Optional[Intention]: """Get an intention by ID.""" with self._lock: return self._intentions.get(intention_id) def update(self, intention_id: str, updates: Dict[str, Any]) -> bool: """ Update an intention's fields. Args: intention_id: ID of intention to update updates: Dictionary of field updates Returns: True if updated successfully """ with self._lock: intention = self._intentions.get(intention_id) if not intention: return False # Remove from indexes before update self._remove_from_indexes(intention) # Apply updates for key, value in updates.items(): if hasattr(intention, key): setattr(intention, key, value) # Re-add to indexes self._update_indexes(intention) logger.debug(f"Updated intention: {intention_id}") return True def remove(self, intention_id: str) -> bool: """Remove an intention from the store.""" with self._lock: intention = self._intentions.get(intention_id) if not intention: return False self._remove_from_indexes(intention) del self._intentions[intention_id] logger.info(f"Removed intention: {intention_id}") return True def get_by_status(self, status: IntentionStatus) -> List[Intention]: """Get all intentions with a specific status.""" with self._lock: ids = self._by_status.get(status, set()) return [self._intentions[id] for id in ids if id in self._intentions] def get_by_priority(self, priority: Priority) -> List[Intention]: """Get all intentions with a specific priority.""" with self._lock: ids = self._by_priority.get(priority, set()) return [self._intentions[id] for id in ids if id in self._intentions] def get_by_cue_type(self, cue_type: CueType) -> List[Intention]: """Get all intentions with a specific cue type.""" with self._lock: ids = self._by_cue_type.get(cue_type, set()) return [self._intentions[id] for id in ids if id in self._intentions] def get_by_tag(self, tag: str) -> List[Intention]: """Get all intentions with a specific tag.""" with self._lock: ids = self._by_tag.get(tag, set()) return [self._intentions[id] for id in ids if id in self._intentions] def get_pending_time_based(self, before_time: Optional[float] = None) -> List[Intention]: """Get pending time-based intentions due before a specific time.""" with self._lock: before_time = before_time or time.time() result = [] for intention_id in self._by_cue_type.get(CueType.TIME_BASED, set()): intention = self._intentions.get(intention_id) if not intention: continue if intention.status == IntentionStatus.PENDING: if intention.trigger_time and intention.trigger_time <= before_time: result.append(intention) # Sort by trigger time result.sort(key=lambda x: x.trigger_time or 0) return result def get_active_event_based(self) -> List[Intention]: """Get all active event-based intentions.""" with self._lock: result = [] for intention_id in self._by_cue_type.get(CueType.EVENT_BASED, set()): intention = self._intentions.get(intention_id) if intention and intention.status in (IntentionStatus.PENDING, IntentionStatus.ACTIVE): result.append(intention) # Sort by priority result.sort(key=lambda x: x.priority.value) return result def cleanup_expired(self) -> int: """ Clean up expired intentions. Returns: Number of intentions marked as expired """ count = 0 with self._lock: for intention in list(self._intentions.values()): if intention.is_expired() and intention.status == IntentionStatus.PENDING: self._remove_from_indexes(intention) intention.status = IntentionStatus.EXPIRED self._update_indexes(intention) self._stats["total_expired"] += 1 count += 1 if count > 0: logger.info(f"Cleaned up {count} expired intentions") return count def get_stats(self) -> Dict[str, Any]: """Get store statistics.""" with self._lock: return { **self._stats, "current_total": len(self._intentions), "by_status": {s.name: len(ids) for s, ids in self._by_status.items()}, "by_priority": {p.name: len(ids) for p, ids in self._by_priority.items()}, "by_cue_type": {c.name: len(ids) for c, ids in self._by_cue_type.items()} } def _update_indexes(self, intention: Intention) -> None: """Update all indexes for an intention.""" self._by_status[intention.status].add(intention.id) self._by_priority[intention.priority].add(intention.id) self._by_cue_type[intention.cue_type].add(intention.id) for tag in intention.tags: self._by_tag[tag].add(intention.id) def _remove_from_indexes(self, intention: Intention) -> None: """Remove an intention from all indexes.""" self._by_status[intention.status].discard(intention.id) self._by_priority[intention.priority].discard(intention.id) self._by_cue_type[intention.cue_type].discard(intention.id) for tag in intention.tags: self._by_tag[tag].discard(intention.id) def persist(self) -> None: """Persist all intentions to disk.""" if not self.storage_path: return with self._lock: data = { "intentions": [i.to_dict() for i in self._intentions.values()], "stats": self._stats, "persisted_at": time.time() } # Ensure directory exists self.storage_path.parent.mkdir(parents=True, exist_ok=True) # Atomic write temp_path = self.storage_path.with_suffix('.tmp') with open(temp_path, 'w') as f: json.dump(data, f, indent=2) temp_path.replace(self.storage_path) logger.debug(f"Persisted {len(self._intentions)} intentions to disk") def _load_from_disk(self) -> None: """Load intentions from disk.""" try: with open(self.storage_path, 'r') as f: data = json.load(f) for intention_data in data.get("intentions", []): intention = Intention.from_dict(intention_data) self._intentions[intention.id] = intention self._update_indexes(intention) self._stats.update(data.get("stats", {})) logger.info(f"Loaded {len(self._intentions)} intentions from disk") except Exception as e: logger.error(f"Failed to load intentions: {e}") def _start_persist_thread(self) -> None: """Start the auto-persistence thread.""" self._running = True self._persist_thread = threading.Thread( target=self._persist_loop, daemon=True, name="IntentionPersist" ) self._persist_thread.start() def _persist_loop(self) -> None: """Main persistence loop.""" while self._running: time.sleep(self.persist_interval) if self._running: try: self.persist() except Exception as e: logger.error(f"Persistence error: {e}") def shutdown(self) -> None: """Shutdown the store gracefully.""" self._running = False if self._persist_thread: self._persist_thread.join(timeout=5.0) self.persist() logger.info("IntentionStore shutdown complete") # ============================================================================= # EVENT-BASED CUE # ============================================================================= class EventBasedCue: """ Triggers intentions based on contextual events. Features: - Pattern matching on event properties - Semantic similarity matching - Composite conditions (AND, OR, NOT) - Confidence scoring """ def __init__(self, intention: Intention): """ Initialize event-based cue. Args: intention: The intention this cue is associated with """ self.intention = intention self.conditions = intention.cue_conditions self._compiled_patterns: Dict[str, re.Pattern] = {} self._compile_patterns() def _compile_patterns(self) -> None: """Pre-compile regex patterns from conditions.""" for key, condition in self.conditions.items(): if isinstance(condition, dict) and "pattern" in condition: try: self._compiled_patterns[key] = re.compile( condition["pattern"], re.IGNORECASE if condition.get("ignore_case", True) else 0 ) except re.error as e: logger.warning(f"Invalid regex pattern for {key}: {e}") def evaluate(self, event: Dict[str, Any]) -> CueMatch: """ Evaluate if an event matches this cue's conditions. Args: event: Event data to evaluate Returns: CueMatch with score and matched conditions """ matched = {} total_weight = 0 matched_weight = 0 for key, condition in self.conditions.items(): weight = 1.0 if isinstance(condition, dict): weight = condition.get("weight", 1.0) match_result = self._evaluate_condition(key, condition, event) else: # Simple equality check match_result = event.get(key) == condition total_weight += weight if match_result: matched[key] = event.get(key) matched_weight += weight score = matched_weight / total_weight if total_weight > 0 else 0.0 return CueMatch( intention_id=self.intention.id, match_score=score, matched_conditions=matched ) def _evaluate_condition( self, key: str, condition: Dict[str, Any], event: Dict[str, Any] ) -> bool: """Evaluate a single condition.""" event_value = event.get(key) # Pattern matching if "pattern" in condition: pattern = self._compiled_patterns.get(key) if pattern and event_value: return bool(pattern.search(str(event_value))) return False # Equality if "equals" in condition: return event_value == condition["equals"] # Contains if "contains" in condition: if isinstance(event_value, str): return condition["contains"].lower() in event_value.lower() if isinstance(event_value, (list, set)): return condition["contains"] in event_value return False # Range check if "min" in condition or "max" in condition: if event_value is None: return False min_val = condition.get("min", float("-inf")) max_val = condition.get("max", float("inf")) return min_val <= event_value <= max_val # Exists check if "exists" in condition: exists = event_value is not None return exists == condition["exists"] # In list if "in" in condition: return event_value in condition["in"] # NOT condition if "not" in condition: return not self._evaluate_condition(key, condition["not"], event) # AND conditions if "and" in condition: return all( self._evaluate_condition(key, c, event) for c in condition["and"] ) # OR conditions if "or" in condition: return any( self._evaluate_condition(key, c, event) for c in condition["or"] ) return False def get_required_keys(self) -> Set[str]: """Get the set of event keys this cue requires.""" return set(self.conditions.keys()) # ============================================================================= # TIME-BASED CUE # ============================================================================= class TimeBasedCue: """ Triggers intentions at specific times. Features: - Absolute time triggers - Relative time triggers (in X minutes) - Recurring schedules (cron-like) - Time window matching """ # Day of week constants MONDAY = 0 TUESDAY = 1 WEDNESDAY = 2 THURSDAY = 3 FRIDAY = 4 SATURDAY = 5 SUNDAY = 6 def __init__(self, intention: Intention): """ Initialize time-based cue. Args: intention: The intention this cue is associated with """ self.intention = intention self.trigger_time = intention.trigger_time self.conditions = intention.cue_conditions # Parse recurring schedule if present self.is_recurring = self.conditions.get("recurring", False) self.recurrence_pattern = self.conditions.get("pattern", None) self.days_of_week = self.conditions.get("days_of_week", None) self.time_of_day = self.conditions.get("time_of_day", None) self.interval_seconds = self.conditions.get("interval_seconds", None) self.last_trigger = intention.last_execution def should_trigger(self, current_time: Optional[float] = None) -> bool: """ Check if this cue should trigger now. Args: current_time: Current timestamp (defaults to now) Returns: True if cue should trigger """ current_time = current_time or time.time() # Check if already executed for non-recurring if not self.is_recurring: if self.intention.status in (IntentionStatus.EXECUTED, IntentionStatus.TRIGGERED): return False # Absolute time trigger if self.trigger_time and not self.is_recurring: return current_time >= self.trigger_time # Recurring triggers if self.is_recurring: return self._check_recurring(current_time) return False def _check_recurring(self, current_time: float) -> bool: """Check recurring trigger conditions.""" now = datetime.fromtimestamp(current_time) # Interval-based recurrence if self.interval_seconds: if self.last_trigger is None: return True return (current_time - self.last_trigger) >= self.interval_seconds # Day of week + time of day if self.days_of_week is not None and self.time_of_day: if now.weekday() not in self.days_of_week: return False hour, minute = self.time_of_day target_time = now.replace(hour=hour, minute=minute, second=0, microsecond=0) # Check if we're within a 1-minute window of the target time if abs((now - target_time).total_seconds()) <= 60: # Ensure we haven't already triggered today if self.last_trigger: last = datetime.fromtimestamp(self.last_trigger) if last.date() == now.date(): return False return True # Time window if "time_window" in self.conditions: window = self.conditions["time_window"] start_hour, start_minute = window.get("start", (0, 0)) end_hour, end_minute = window.get("end", (23, 59)) start = now.replace(hour=start_hour, minute=start_minute) end = now.replace(hour=end_hour, minute=end_minute) return start <= now <= end return False def get_next_trigger_time(self) -> Optional[float]: """Calculate the next trigger time for this cue.""" now = time.time() if not self.is_recurring and self.trigger_time: if self.trigger_time > now: return self.trigger_time return None if self.is_recurring: if self.interval_seconds: last = self.last_trigger or now return last + self.interval_seconds if self.days_of_week and self.time_of_day: return self._calculate_next_scheduled_time() return None def _calculate_next_scheduled_time(self) -> float: """Calculate next scheduled time for day-of-week recurrence.""" now = datetime.now() hour, minute = self.time_of_day # Check each day starting from today for day_offset in range(8): # Up to 7 days ahead check_date = now + timedelta(days=day_offset) if check_date.weekday() in self.days_of_week: target = check_date.replace( hour=hour, minute=minute, second=0, microsecond=0 ) if target > now: return target.timestamp() return now.timestamp() + 86400 # Fallback: 24 hours # ============================================================================= # MONITORING PROCESS # ============================================================================= class MonitoringProcess: """ Background monitoring for cue detection. Features: - Continuous event stream monitoring - Time-based trigger checking - Configurable monitoring intervals - Event buffering and batch processing """ def __init__( self, intention_store: IntentionStore, check_interval: float = 1.0, event_buffer_size: int = 100 ): """ Initialize the monitoring process. Args: intention_store: The intention store to monitor check_interval: Seconds between monitoring cycles event_buffer_size: Maximum events to buffer """ self.intention_store = intention_store self.check_interval = check_interval self.event_buffer_size = event_buffer_size # Event buffer self._event_buffer: List[Dict[str, Any]] = [] self._buffer_lock = threading.Lock() # Cue caches self._event_cues: Dict[str, EventBasedCue] = {} self._time_cues: Dict[str, TimeBasedCue] = {} # Callbacks self._on_trigger: Optional[Callable[[Intention, CueMatch], None]] = None self._on_time_trigger: Optional[Callable[[Intention], None]] = None # Threading self._running = False self._monitor_thread: Optional[threading.Thread] = None # Stats self._stats = { "events_processed": 0, "time_checks": 0, "triggers_fired": 0 } def set_trigger_callback( self, on_event: Optional[Callable[[Intention, CueMatch], None]] = None, on_time: Optional[Callable[[Intention], None]] = None ) -> None: """Set callbacks for trigger events.""" self._on_trigger = on_event self._on_time_trigger = on_time def submit_event(self, event: Dict[str, Any]) -> None: """ Submit an event for monitoring. Args: event: Event data to check against cues """ with self._buffer_lock: if len(self._event_buffer) >= self.event_buffer_size: self._event_buffer.pop(0) # Remove oldest event["_submitted_at"] = time.time() self._event_buffer.append(event) def start(self) -> None: """Start the monitoring process.""" if self._running: return self._running = True self._refresh_cues() self._monitor_thread = threading.Thread( target=self._monitor_loop, daemon=True, name="ProspectiveMonitor" ) self._monitor_thread.start() logger.info("Monitoring process started") def stop(self) -> None: """Stop the monitoring process.""" self._running = False if self._monitor_thread: self._monitor_thread.join(timeout=5.0) logger.info("Monitoring process stopped") def _monitor_loop(self) -> None: """Main monitoring loop.""" last_cue_refresh = time.time() cue_refresh_interval = 30.0 # Refresh cues every 30 seconds while self._running: try: # Refresh cues periodically if time.time() - last_cue_refresh > cue_refresh_interval: self._refresh_cues() last_cue_refresh = time.time() # Process events self._process_event_buffer() # Check time-based triggers self._check_time_triggers() except Exception as e: logger.error(f"Monitor loop error: {e}") time.sleep(self.check_interval) def _refresh_cues(self) -> None: """Refresh cue caches from intention store.""" # Event-based cues event_intentions = self.intention_store.get_active_event_based() self._event_cues = { i.id: EventBasedCue(i) for i in event_intentions } # Time-based cues time_intentions = self.intention_store.get_by_cue_type(CueType.TIME_BASED) self._time_cues = { i.id: TimeBasedCue(i) for i in time_intentions if i.status in (IntentionStatus.PENDING, IntentionStatus.ACTIVE) } logger.debug(f"Refreshed cues: {len(self._event_cues)} event, {len(self._time_cues)} time") def _process_event_buffer(self) -> None: """Process buffered events.""" with self._buffer_lock: events = self._event_buffer.copy() self._event_buffer.clear() for event in events: self._stats["events_processed"] += 1 self._check_event_triggers(event) def _check_event_triggers(self, event: Dict[str, Any]) -> None: """Check if an event triggers any cues.""" for intention_id, cue in self._event_cues.items(): match = cue.evaluate(event) # Check if match score meets threshold threshold = cue.intention.cue_conditions.get("match_threshold", 0.8) if match.match_score >= threshold: self._fire_event_trigger(cue.intention, match) def _check_time_triggers(self) -> None: """Check time-based triggers.""" self._stats["time_checks"] += 1 current_time = time.time() for intention_id, cue in list(self._time_cues.items()): if cue.should_trigger(current_time): self._fire_time_trigger(cue.intention) def _fire_event_trigger(self, intention: Intention, match: CueMatch) -> None: """Fire an event-based trigger.""" self._stats["triggers_fired"] += 1 # Update intention status self.intention_store.update(intention.id, { "status": IntentionStatus.TRIGGERED }) logger.info( f"Event trigger fired: {intention.id} " f"(score: {match.match_score:.2f})" ) if self._on_trigger: try: self._on_trigger(intention, match) except Exception as e: logger.error(f"Trigger callback error: {e}") def _fire_time_trigger(self, intention: Intention) -> None: """Fire a time-based trigger.""" self._stats["triggers_fired"] += 1 # Update intention updates = {"status": IntentionStatus.TRIGGERED} # For recurring, update last_execution cue = self._time_cues.get(intention.id) if cue and cue.is_recurring: updates["last_execution"] = time.time() self.intention_store.update(intention.id, updates) logger.info(f"Time trigger fired: {intention.id}") if self._on_time_trigger: try: self._on_time_trigger(intention) except Exception as e: logger.error(f"Time trigger callback error: {e}") def get_stats(self) -> Dict[str, Any]: """Get monitoring statistics.""" return { **self._stats, "active_event_cues": len(self._event_cues), "active_time_cues": len(self._time_cues), "buffered_events": len(self._event_buffer) } # ============================================================================= # RETRIEVAL PROCESS # ============================================================================= class RetrievalProcess: """ Intelligent intention retrieval. Features: - Context-aware retrieval - Semantic similarity matching - Priority-based ranking - Recency weighting """ def __init__(self, intention_store: IntentionStore): """ Initialize the retrieval process. Args: intention_store: The intention store to retrieve from """ self.intention_store = intention_store # Retrieval weights self.priority_weight = 0.3 self.recency_weight = 0.2 self.relevance_weight = 0.5 def retrieve_relevant( self, context: Dict[str, Any], limit: int = 10, min_relevance: float = 0.3 ) -> List[Tuple[Intention, float]]: """ Retrieve intentions relevant to the current context. Args: context: Current context information limit: Maximum number of intentions to return min_relevance: Minimum relevance score threshold Returns: List of (intention, score) tuples sorted by relevance """ scored = [] # Get active intentions for status in (IntentionStatus.PENDING, IntentionStatus.ACTIVE): for intention in self.intention_store.get_by_status(status): score = self._calculate_relevance(intention, context) if score >= min_relevance: scored.append((intention, score)) # Sort by score descending scored.sort(key=lambda x: x[1], reverse=True) return scored[:limit] def retrieve_by_action(self, action_name: str) -> List[Intention]: """Retrieve intentions by action name.""" result = [] for status in (IntentionStatus.PENDING, IntentionStatus.ACTIVE): for intention in self.intention_store.get_by_status(status): if intention.action == action_name: result.append(intention) return result def retrieve_upcoming( self, time_window_seconds: float = 3600 ) -> List[Intention]: """ Retrieve upcoming time-based intentions. Args: time_window_seconds: Look-ahead window in seconds Returns: List of upcoming intentions sorted by trigger time """ deadline = time.time() + time_window_seconds return self.intention_store.get_pending_time_based(before_time=deadline) def retrieve_by_similarity( self, query: str, limit: int = 5 ) -> List[Tuple[Intention, float]]: """ Retrieve intentions by content similarity. Args: query: Search query limit: Maximum results Returns: List of (intention, similarity) tuples """ scored = [] query_lower = query.lower() query_words = set(query_lower.split()) for status in (IntentionStatus.PENDING, IntentionStatus.ACTIVE): for intention in self.intention_store.get_by_status(status): similarity = self._calculate_text_similarity( query_words, intention.content.lower() ) if similarity > 0: scored.append((intention, similarity)) scored.sort(key=lambda x: x[1], reverse=True) return scored[:limit] def _calculate_relevance( self, intention: Intention, context: Dict[str, Any] ) -> float: """Calculate overall relevance score.""" scores = [] # Priority score (1.0 for CRITICAL, decreasing) priority_score = (6 - intention.priority.value) / 5 scores.append(self.priority_weight * priority_score) # Recency score (decay based on age) age_hours = (time.time() - intention.created_at) / 3600 recency_score = max(0, 1 - (age_hours / 168)) # Decay over 1 week scores.append(self.recency_weight * recency_score) # Context relevance relevance_score = self._calculate_context_match(intention, context) scores.append(self.relevance_weight * relevance_score) return sum(scores) def _calculate_context_match( self, intention: Intention, context: Dict[str, Any] ) -> float: """Calculate context match score.""" if not intention.cue_conditions: return 0.5 # Default neutral score matches = 0 total = len(intention.cue_conditions) for key, condition in intention.cue_conditions.items(): if key in context: if isinstance(condition, dict): # Complex condition if "equals" in condition: if context[key] == condition["equals"]: matches += 1 elif "contains" in condition: if condition["contains"] in str(context[key]): matches += 1 else: # Simple equality if context[key] == condition: matches += 1 return matches / total if total > 0 else 0 def _calculate_text_similarity( self, query_words: Set[str], content: str ) -> float: """Calculate simple text similarity using word overlap.""" content_words = set(content.split()) if not query_words: return 0.0 intersection = query_words & content_words return len(intersection) / len(query_words) # ============================================================================= # EXECUTION TRACKER # ============================================================================= class ExecutionTracker: """ Track and verify intention execution. Features: - Execution history logging - Success/failure tracking - Retry management - Performance metrics """ def __init__( self, intention_store: IntentionStore, max_retries: int = 3, retry_delay_seconds: float = 60.0, history_limit: int = 1000 ): """ Initialize the execution tracker. Args: intention_store: The intention store max_retries: Maximum retry attempts for failed executions retry_delay_seconds: Delay between retries history_limit: Maximum execution records to keep """ self.intention_store = intention_store self.max_retries = max_retries self.retry_delay_seconds = retry_delay_seconds self.history_limit = history_limit # Execution history self._history: List[ExecutionRecord] = [] self._by_intention: Dict[str, List[ExecutionRecord]] = defaultdict(list) # Action registry self._actions: Dict[str, Callable] = {} # Retry queue self._retry_queue: List[Tuple[str, float]] = [] # (intention_id, retry_after) # Stats self._stats = { "total_executions": 0, "successful": 0, "failed": 0, "retries": 0, "total_duration_ms": 0.0 } # Lock self._lock = threading.Lock() def register_action(self, name: str, handler: Callable) -> None: """ Register an action handler. Args: name: Action name handler: Callable that implements the action """ self._actions[name] = handler logger.info(f"Registered action: {name}") def execute( self, intention: Intention, context: Optional[Dict[str, Any]] = None ) -> ExecutionRecord: """ Execute an intention. Args: intention: The intention to execute context: Additional execution context Returns: ExecutionRecord with the result """ context = context or {} start_time = time.time() record = ExecutionRecord( intention_id=intention.id, context=context ) try: # Get the action handler handler = self._actions.get(intention.action) if not handler: raise ValueError(f"Unknown action: {intention.action}") # Execute the action output = handler(**intention.action_params, _context=context) # Success record.result = ExecutionResult.SUCCESS record.output = output # Update intention self.intention_store.update(intention.id, { "status": IntentionStatus.EXECUTED, "execution_count": intention.execution_count + 1, "last_execution": time.time() }) self._stats["successful"] += 1 except Exception as e: # Failure record.result = ExecutionResult.FAILED record.error = str(e) # Check if we should retry if intention.execution_count < self.max_retries: self._schedule_retry(intention.id) self._stats["retries"] += 1 self.intention_store.update(intention.id, { "status": IntentionStatus.PENDING, "execution_count": intention.execution_count + 1 }) else: self.intention_store.update(intention.id, { "status": IntentionStatus.FAILED, "execution_count": intention.execution_count + 1 }) self._stats["failed"] += 1 logger.error(f"Execution failed for {intention.id}: {e}") # Record timing record.duration_ms = (time.time() - start_time) * 1000 # Store record self._record_execution(record) return record def execute_by_id( self, intention_id: str, context: Optional[Dict[str, Any]] = None ) -> Optional[ExecutionRecord]: """Execute an intention by ID.""" intention = self.intention_store.get(intention_id) if not intention: logger.warning(f"Intention not found: {intention_id}") return None return self.execute(intention, context) def get_execution_history( self, intention_id: Optional[str] = None, limit: int = 100 ) -> List[ExecutionRecord]: """ Get execution history. Args: intention_id: Filter by intention ID limit: Maximum records to return Returns: List of execution records """ with self._lock: if intention_id: records = self._by_intention.get(intention_id, []) else: records = self._history return list(reversed(records[-limit:])) def get_pending_retries(self) -> List[str]: """Get intention IDs pending retry.""" current_time = time.time() ready = [] with self._lock: remaining = [] for intention_id, retry_after in self._retry_queue: if current_time >= retry_after: ready.append(intention_id) else: remaining.append((intention_id, retry_after)) self._retry_queue = remaining return ready def process_retries(self) -> List[ExecutionRecord]: """Process pending retries.""" results = [] for intention_id in self.get_pending_retries(): result = self.execute_by_id(intention_id) if result: results.append(result) return results def get_stats(self) -> Dict[str, Any]: """Get execution statistics.""" with self._lock: avg_duration = ( self._stats["total_duration_ms"] / self._stats["total_executions"] if self._stats["total_executions"] > 0 else 0 ) return { **self._stats, "average_duration_ms": avg_duration, "success_rate": ( self._stats["successful"] / self._stats["total_executions"] if self._stats["total_executions"] > 0 else 0 ), "pending_retries": len(self._retry_queue), "history_size": len(self._history) } def _record_execution(self, record: ExecutionRecord) -> None: """Record an execution.""" with self._lock: self._history.append(record) self._by_intention[record.intention_id].append(record) self._stats["total_executions"] += 1 self._stats["total_duration_ms"] += record.duration_ms # Trim history if needed if len(self._history) > self.history_limit: old_record = self._history.pop(0) if old_record.intention_id in self._by_intention: intention_history = self._by_intention[old_record.intention_id] if old_record in intention_history: intention_history.remove(old_record) def _schedule_retry(self, intention_id: str) -> None: """Schedule an intention for retry.""" retry_time = time.time() + self.retry_delay_seconds with self._lock: self._retry_queue.append((intention_id, retry_time)) # ============================================================================= # PROSPECTIVE MEMORY SYSTEM (UNIFIED) # ============================================================================= class ProspectiveMemorySystem: """ Unified prospective memory system for AIVA. Combines all components into a cohesive system for managing future intentions with event-based and time-based triggering. """ def __init__( self, storage_path: Optional[str] = None, check_interval: float = 1.0, auto_start: bool = True ): """ Initialize the prospective memory system. Args: storage_path: Path for persistent storage check_interval: Monitoring check interval auto_start: Automatically start monitoring """ # Initialize components self.intention_store = IntentionStore( storage_path=storage_path, auto_persist=True ) self.monitoring = MonitoringProcess( intention_store=self.intention_store, check_interval=check_interval ) self.retrieval = RetrievalProcess( intention_store=self.intention_store ) self.execution = ExecutionTracker( intention_store=self.intention_store ) # Wire up callbacks self.monitoring.set_trigger_callback( on_event=self._on_event_trigger, on_time=self._on_time_trigger ) # Start if requested if auto_start: self.start() def create_intention( self, content: str, action: str, cue_type: CueType = CueType.EVENT_BASED, trigger_time: Optional[datetime] = None, trigger_delay: Optional[timedelta] = None, cue_conditions: Optional[Dict[str, Any]] = None, priority: Priority = Priority.MEDIUM, expiry: Optional[datetime] = None, tags: Optional[Set[str]] = None, action_params: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None ) -> str: """ Create a new intention. Args: content: Description of the intention action: Action to execute cue_type: Type of triggering cue trigger_time: Absolute trigger time (for time-based) trigger_delay: Relative trigger delay (for time-based) cue_conditions: Conditions for triggering priority: Intention priority expiry: When the intention expires tags: Tags for categorization action_params: Parameters for the action metadata: Additional metadata Returns: Intention ID """ # Calculate trigger time trigger_ts = None if cue_type == CueType.TIME_BASED: if trigger_time: trigger_ts = trigger_time.timestamp() elif trigger_delay: trigger_ts = time.time() + trigger_delay.total_seconds() # Calculate expiry expiry_ts = expiry.timestamp() if expiry else None intention = Intention( content=content, action=action, action_params=action_params or {}, cue_type=cue_type, cue_conditions=cue_conditions or {}, trigger_time=trigger_ts, expiry_time=expiry_ts, priority=priority, tags=tags or set(), metadata=metadata or {} ) return self.intention_store.store(intention) def remind_in( self, content: str, action: str, delay: timedelta, priority: Priority = Priority.MEDIUM, action_params: Optional[Dict[str, Any]] = None ) -> str: """ Create a reminder to execute in a specific time. Args: content: Reminder description action: Action to execute delay: Time delay from now priority: Reminder priority action_params: Action parameters Returns: Intention ID """ return self.create_intention( content=content, action=action, cue_type=CueType.TIME_BASED, trigger_delay=delay, priority=priority, action_params=action_params ) def remind_at( self, content: str, action: str, trigger_time: datetime, priority: Priority = Priority.MEDIUM, action_params: Optional[Dict[str, Any]] = None ) -> str: """ Create a reminder at a specific time. Args: content: Reminder description action: Action to execute trigger_time: When to trigger priority: Reminder priority action_params: Action parameters Returns: Intention ID """ return self.create_intention( content=content, action=action, cue_type=CueType.TIME_BASED, trigger_time=trigger_time, priority=priority, action_params=action_params ) def when_event( self, content: str, action: str, conditions: Dict[str, Any], priority: Priority = Priority.MEDIUM, action_params: Optional[Dict[str, Any]] = None, match_threshold: float = 0.8 ) -> str: """ Create an intention triggered by an event. Args: content: Intention description action: Action to execute conditions: Event conditions to match priority: Intention priority action_params: Action parameters match_threshold: Minimum match score to trigger Returns: Intention ID """ conditions["match_threshold"] = match_threshold return self.create_intention( content=content, action=action, cue_type=CueType.EVENT_BASED, cue_conditions=conditions, priority=priority, action_params=action_params ) def submit_event(self, event: Dict[str, Any]) -> None: """Submit an event for monitoring.""" self.monitoring.submit_event(event) def register_action(self, name: str, handler: Callable) -> None: """Register an action handler.""" self.execution.register_action(name, handler) def get_upcoming(self, hours: float = 24) -> List[Intention]: """Get upcoming intentions within a time window.""" seconds = hours * 3600 return self.retrieval.retrieve_upcoming(seconds) def get_relevant( self, context: Dict[str, Any], limit: int = 10 ) -> List[Tuple[Intention, float]]: """Get relevant intentions for a context.""" return self.retrieval.retrieve_relevant(context, limit) def cancel(self, intention_id: str) -> bool: """Cancel an intention.""" return self.intention_store.update(intention_id, { "status": IntentionStatus.CANCELLED }) def get_stats(self) -> Dict[str, Any]: """Get comprehensive system statistics.""" return { "store": self.intention_store.get_stats(), "monitoring": self.monitoring.get_stats(), "execution": self.execution.get_stats() } def start(self) -> None: """Start the monitoring process.""" self.monitoring.start() def stop(self) -> None: """Stop the system.""" self.monitoring.stop() self.intention_store.shutdown() def _on_event_trigger(self, intention: Intention, match: CueMatch) -> None: """Handle event-based triggers.""" logger.info( f"Event trigger: {intention.content[:50]} " f"(match: {match.match_score:.2f})" ) self.execution.execute(intention, {"trigger_match": match.matched_conditions}) def _on_time_trigger(self, intention: Intention) -> None: """Handle time-based triggers.""" logger.info(f"Time trigger: {intention.content[:50]}") self.execution.execute(intention) # ============================================================================= # EXAMPLE USAGE AND TESTS # ============================================================================= def example_action(message: str, _context: Dict = None) -> str: """Example action handler.""" print(f"[ACTION] Executing: {message}") return f"Executed: {message}" def recurring_action(task: str, _context: Dict = None) -> str: """Recurring task action.""" print(f"[RECURRING] {task} at {datetime.now()}") return f"Completed: {task}" if __name__ == "__main__": print("=" * 60) print("AIVA Prospective Memory System - Test Suite") print("=" * 60) # Initialize system system = ProspectiveMemorySystem( storage_path="/tmp/aiva_prospective_memory.json", check_interval=0.5, auto_start=True ) # Register actions system.register_action("example", example_action) system.register_action("recurring", recurring_action) # Test 1: Time-based reminder print("\n[TEST 1] Creating time-based reminder (5 seconds)...") reminder_id = system.remind_in( content="Check system status", action="example", delay=timedelta(seconds=5), priority=Priority.HIGH, action_params={"message": "System check complete!"} ) print(f"Created reminder: {reminder_id}") # Test 2: Event-based intention print("\n[TEST 2] Creating event-based intention...") event_id = system.when_event( content="Process high-priority lead", action="example", conditions={ "event_type": {"equals": "new_lead"}, "priority": {"equals": "high"} }, action_params={"message": "Processing lead..."} ) print(f"Created event intention: {event_id}") # Test 3: Submit matching event print("\n[TEST 3] Submitting matching event...") system.submit_event({ "event_type": "new_lead", "priority": "high", "lead_id": "12345" }) # Test 4: Get upcoming intentions print("\n[TEST 4] Getting upcoming intentions...") upcoming = system.get_upcoming(hours=1) print(f"Upcoming intentions: {len(upcoming)}") for intention in upcoming: print(f" - {intention.content} @ {intention.trigger_time}") # Test 5: Get stats print("\n[TEST 5] System statistics...") stats = system.get_stats() print(f"Store stats: {stats['store']}") print(f"Monitoring stats: {stats['monitoring']}") print(f"Execution stats: {stats['execution']}") # Wait for triggers print("\n[WAITING] Waiting for triggers (10 seconds)...") try: time.sleep(10) except KeyboardInterrupt: pass # Final stats print("\n[FINAL] Final statistics...") final_stats = system.get_stats() print(f"Total executions: {final_stats['execution']['total_executions']}") print(f"Successful: {final_stats['execution']['successful']}") print(f"Failed: {final_stats['execution']['failed']}") # Cleanup print("\n[CLEANUP] Shutting down...") system.stop() print("\n" + "=" * 60) print("Test complete!") print("=" * 60)