#!/usr/bin/env python3 """ Steward Profile System - Learning About MIRA's User(s) ====================================================== This module implements a comprehensive profile system that passively learns about the person or team using MIRA. It builds a rich understanding of the steward through conversation analysis, code patterns, and behavioral observation. The system supports both individual and team profiles, automatically detecting when multiple people are involved and building a collective understanding. """ import json import os from datetime import datetime, timedelta from typing import Dict, List, Optional, Set, Any, Tuple from dataclasses import dataclass, field, asdict from collections import defaultdict, Counter import re from pathlib import Path from core.mira_path_resolver import get_mira_memory_dir @dataclass class CommunicationStyle: """Tracks how the steward communicates""" formality_level: float = 0.5 # 0=casual, 1=formal technical_depth: float = 0.5 # 0=basic, 1=expert preferred_greetings: List[str] = field(default_factory=list) common_phrases: List[str] = field(default_factory=list) emoji_usage: float = 0.0 # Frequency of emoji use code_in_messages: float = 0.0 # How often they include code question_patterns: List[str] = field(default_factory=list) @dataclass class TechnicalProfile: """Technical skills and preferences""" primary_languages: List[str] = field(default_factory=list) frameworks: List[str] = field(default_factory=list) tools: List[str] = field(default_factory=list) expertise_areas: List[str] = field(default_factory=list) coding_patterns: Dict[str, float] = field(default_factory=dict) preferred_architectures: List[str] = field(default_factory=list) testing_approach: str = "unknown" documentation_style: str = "unknown" @dataclass class WorkPatterns: """When and how the steward works""" active_hours: Dict[int, float] = field(default_factory=dict) # Hour -> activity level active_days: Dict[str, float] = field(default_factory=dict) # Day -> activity level session_durations: List[float] = field(default_factory=list) break_patterns: List[float] = field(default_factory=list) multitasking_score: float = 0.0 focus_score: float = 0.0 @dataclass class ProjectContext: """What the steward is working on""" project_types: List[str] = field(default_factory=list) current_goals: List[str] = field(default_factory=list) pain_points: List[str] = field(default_factory=list) interests: List[str] = field(default_factory=list) values: List[str] = field(default_factory=list) # What they care about @dataclass class TeamMember: """Individual team member profile""" name: str aliases: List[str] = field(default_factory=list) role: Optional[str] = None first_seen: datetime = field(default_factory=datetime.now) last_seen: datetime = field(default_factory=datetime.now) interaction_count: int = 0 communication_style: CommunicationStyle = field(default_factory=CommunicationStyle) technical_profile: TechnicalProfile = field(default_factory=TechnicalProfile) @dataclass class StewardProfile: """Complete profile of MIRA's steward(s)""" # Identity profile_id: str = field(default_factory=lambda: datetime.now().isoformat()) is_team: bool = False primary_name: Optional[str] = None aliases: List[str] = field(default_factory=list) pronouns: Optional[str] = None # Team information (if applicable) team_members: Dict[str, TeamMember] = field(default_factory=dict) team_size: int = 1 # Communication communication_style: CommunicationStyle = field(default_factory=CommunicationStyle) # Technical technical_profile: TechnicalProfile = field(default_factory=TechnicalProfile) # Work patterns work_patterns: WorkPatterns = field(default_factory=WorkPatterns) # Context project_context: ProjectContext = field(default_factory=ProjectContext) # Relationship with MIRA relationship_quality: float = 0.5 # 0=frustrated, 1=delighted trust_level: float = 0.5 # 0=skeptical, 1=trusting engagement_level: float = 0.5 # 0=passive, 1=highly engaged # Metadata created_at: datetime = field(default_factory=datetime.now) last_updated: datetime = field(default_factory=datetime.now) total_interactions: int = 0 # Key insights summary: str = "" key_traits: List[str] = field(default_factory=list) needs: List[str] = field(default_factory=list) preferences: List[str] = field(default_factory=list) class StewardProfileManager: """Manages the steward profile with persistence and learning""" def __init__(self): self.memory_dir = get_mira_memory_dir() self.profile_dir = os.path.join(self.memory_dir, "steward_profile") self.profile_file = os.path.join(self.profile_dir, "profile.json") self.learning_file = os.path.join(self.profile_dir, "learning_data.jsonl") os.makedirs(self.profile_dir, exist_ok=True) self.profile = self._load_profile() # Pattern matchers - flexible for individuals and teams self.name_patterns = [ # Individual patterns with titles r"(?:I'm|I am|my (?:full )?name is|call me)\s+((?:Dr\.?|Mr\.?|Ms\.?|Mrs\.?|Prof\.?|Mx\.?)?\s*(?:[\w'-]+(?:\s+[\w'-]+){0,3}))(?:\s*[,.]|\s+and\s|\s+from\s|\s+working\s|$)", r"(?:This is|It's)\s+((?:Dr\.?|Mr\.?|Ms\.?|Mrs\.?|Prof\.?|Mx\.?)?\s*(?:[\w'-]+(?:\s+[\w'-]+){0,3}))(?:\s+here|\s+from\s|[,.]|$)", r"^((?:Dr\.?|Mr\.?|Ms\.?|Mrs\.?|Prof\.?|Mx\.?)?\s*(?:[\w'-]+(?:\s+[\w'-]+){0,3}))\s+here[,.]", r"(?:Hi|Hello),?\s+I'?m?\s+((?:Dr\.?|Mr\.?|Ms\.?|Mrs\.?|Prof\.?|Mx\.?)?\s*(?:[\w'-]+(?:\s+[\w'-]+){0,3}))[,.]", # Team patterns r"(?:We are|We're|This is)\s+([\w\s&'-]+(?:Team|Squad|Group|Labs?|Inc|LLC|Corp|Company|Collective|Studio))", r"(?:from|representing|with)\s+([\w\s&'-]+(?:Team|Squad|Group|Labs?|Inc|LLC|Corp|Company|Collective|Studio))", # Sign-offs r"^[-–—]\s*((?:Dr\.?|Mr\.?|Ms\.?|Mrs\.?|Prof\.?|Mx\.?)?\s*(?:[\w'-]+(?:\s+[\w'-]+){0,3}))$", r"(?:Best|Regards|Sincerely|Thanks),?\s*[-–—]?\s*((?:[\w'-]+(?:\s+[\w'-]+){0,3}))$", # International/complex names r"(?:Je suis|Ich bin|Mi nombre es|私は)\s+([\w\s'-]+)", ] self.pronoun_patterns = [ r"(?:my pronouns are|pronouns:?)\s*([\w/]+)", r"(?:he/him|she/her|they/them|xe/xem|ze/zir)", ] self.role_patterns = [ r"(?:I'm a|I am a|work as a|working as a)\s+([\w\s]+?)(?:\.|,|$)", r"(?:developer|engineer|designer|manager|architect|analyst|scientist)", ] def _load_profile(self) -> StewardProfile: """Load existing profile or create new one""" if os.path.exists(self.profile_file): try: with open(self.profile_file, 'r') as f: data = json.load(f) # Convert nested dictionaries to dataclasses if 'communication_style' in data and isinstance(data['communication_style'], dict): data['communication_style'] = CommunicationStyle(**data['communication_style']) if 'technical_profile' in data and isinstance(data['technical_profile'], dict): data['technical_profile'] = TechnicalProfile(**data['technical_profile']) if 'work_patterns' in data and isinstance(data['work_patterns'], dict): data['work_patterns'] = WorkPatterns(**data['work_patterns']) if 'project_context' in data and isinstance(data['project_context'], dict): data['project_context'] = ProjectContext(**data['project_context']) if 'relationship' in data and isinstance(data['relationship'], dict): data['relationship'] = RelationshipMetrics(**data['relationship']) # Convert timestamps for field in ['created_at', 'last_updated']: if field in data and data[field]: data[field] = datetime.fromisoformat(data[field]) return StewardProfile(**data) except Exception as e: print(f"Error loading profile: {e}") return StewardProfile() def save_profile(self): """Persist the profile to disk""" try: # Convert to dict and handle datetime serialization data = asdict(self.profile) # Convert datetime objects to ISO format def convert_dates(obj): if isinstance(obj, datetime): return obj.isoformat() elif isinstance(obj, dict): return {k: convert_dates(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_dates(v) for v in obj] return obj data = convert_dates(data) with open(self.profile_file, 'w') as f: json.dump(data, f, indent=2) # Also save to private memory for quick access self._store_in_private_memory() except Exception as e: print(f"Error saving profile: {e}") def _store_in_private_memory(self): """Store profile in private memory for quick retrieval""" try: from core.engine.encrypted_lightning_vidmem import get_claude_private_memory private_memory = get_claude_private_memory() # Create a summary for private storage summary = { "type": "steward_profile", "primary_name": self.profile.primary_name, "is_team": self.profile.is_team, "team_size": self.profile.team_size, "summary": self.profile.summary, "key_traits": self.profile.key_traits, "last_updated": self.profile.last_updated.isoformat() } memory_id = private_memory.store_private_memory( json.dumps(summary), "steward_profile" ) # Store the memory ID for quick retrieval with open(os.path.join(self.profile_dir, "memory_id.txt"), 'w') as f: f.write(memory_id) except Exception as e: print(f"Could not store in private memory: {e}") def learn_from_message(self, message: str, timestamp: Optional[datetime] = None): """Extract information from a single message""" if not timestamp: timestamp = datetime.now() self.profile.total_interactions += 1 self.profile.last_updated = timestamp # Extract name name = self._extract_name(message) if name: self._update_name(name) # Extract pronouns pronouns = self._extract_pronouns(message) if pronouns and not self.profile.pronouns: self.profile.pronouns = pronouns # Analyze communication style self._analyze_communication_style(message) # Extract technical information self._extract_technical_info(message) # Detect team indicators self._detect_team_indicators(message) # Update work patterns self._update_work_patterns(timestamp) # Extract goals and interests self._extract_goals_and_interests(message) # Log learning data self._log_learning_data({ "timestamp": timestamp.isoformat(), "message_length": len(message), "extracted_name": name, "detected_patterns": self._get_detected_patterns(message) }) # Periodically synthesize insights if self.profile.total_interactions % 10 == 0: self._synthesize_insights() self.save_profile() def _extract_name(self, message: str) -> Optional[str]: """Extract name from message""" for pattern in self.name_patterns: match = re.search(pattern, message, re.IGNORECASE) if match: name = match.group(1).strip() # Clean up the name - remove extra spaces, normalize titles name = re.sub(r'\s+', ' ', name) # Multiple spaces to single name = re.sub(r'\.(?=\s|$)', '', name) # Remove trailing periods from titles return name return None def _extract_pronouns(self, message: str) -> Optional[str]: """Extract pronouns from message""" for pattern in self.pronoun_patterns: match = re.search(pattern, message, re.IGNORECASE) if match: return match.group(0).strip() return None def _update_name(self, name: str): """Update name information intelligently""" if not self.profile.primary_name: self.profile.primary_name = name else: # Check if this is a more complete version of the current name current_name_lower = self.profile.primary_name.lower() new_name_lower = name.lower() # If the new name contains the current name, it's likely more complete if current_name_lower in new_name_lower and len(name) > len(self.profile.primary_name): # Move current primary to aliases if not already there if self.profile.primary_name not in self.profile.aliases: self.profile.aliases.append(self.profile.primary_name) self.profile.primary_name = name # If current name contains the new name, new name might be a nickname elif new_name_lower in current_name_lower: if name not in self.profile.aliases: self.profile.aliases.append(name) # If names are completely different elif new_name_lower != current_name_lower: # Check if new name has a title and current doesn't has_title_new = any(title in new_name_lower for title in ['dr', 'prof', 'mr', 'ms', 'mrs']) has_title_current = any(title in current_name_lower for title in ['dr', 'prof', 'mr', 'ms', 'mrs']) if has_title_new and not has_title_current: # New name with title is likely more formal/complete if self.profile.primary_name not in self.profile.aliases: self.profile.aliases.append(self.profile.primary_name) self.profile.primary_name = name else: # Just add as alias if name not in self.profile.aliases: self.profile.aliases.append(name) def _analyze_communication_style(self, message: str): """Analyze how the user communicates""" style = self.profile.communication_style # Formality analysis formal_indicators = ['please', 'could you', 'would you', 'kindly', 'regards'] casual_indicators = ['hey', 'sup', 'gonna', 'wanna', 'lol', 'btw'] formal_count = sum(1 for ind in formal_indicators if ind in message.lower()) casual_count = sum(1 for ind in casual_indicators if ind in message.lower()) if formal_count > casual_count: style.formality_level = min(1.0, style.formality_level + 0.05) elif casual_count > formal_count: style.formality_level = max(0.0, style.formality_level - 0.05) # Technical depth technical_terms = ['api', 'function', 'class', 'method', 'algorithm', 'architecture', 'implementation', 'interface', 'framework', 'dependency'] tech_count = sum(1 for term in technical_terms if term in message.lower()) if tech_count > 2: style.technical_depth = min(1.0, style.technical_depth + 0.05) # Emoji usage emoji_pattern = re.compile(r'[\U0001F600-\U0001F64F\U0001F300-\U0001F5FF\U0001F680-\U0001F6FF\U0001F1E0-\U0001F1FF]') if emoji_pattern.search(message): style.emoji_usage = min(1.0, style.emoji_usage + 0.05) # Code in messages if '```' in message or re.search(r'[a-zA-Z_]\w*\(\)', message): style.code_in_messages = min(1.0, style.code_in_messages + 0.05) def _extract_technical_info(self, message: str): """Extract technical information from message""" tech = self.profile.technical_profile # Language detection languages = { 'python': r'\b(?:python|pip|django|flask|pytest)\b', 'javascript': r'\b(?:javascript|js|node|npm|react|vue|angular)\b', 'typescript': r'\b(?:typescript|ts|tsc)\b', 'java': r'\b(?:java|maven|gradle|spring)\b', 'go': r'\b(?:golang|go)\b', 'rust': r'\b(?:rust|cargo)\b', } for lang, pattern in languages.items(): if re.search(pattern, message, re.IGNORECASE): if lang not in tech.primary_languages: tech.primary_languages.append(lang) # Framework detection frameworks = { 'react': r'\b(?:react|jsx|hooks|useState)\b', 'vue': r'\b(?:vue|vuex|nuxt)\b', 'django': r'\b(?:django|drf)\b', 'express': r'\b(?:express|middleware)\b', 'fastapi': r'\b(?:fastapi|pydantic)\b', } for fw, pattern in frameworks.items(): if re.search(pattern, message, re.IGNORECASE): if fw not in tech.frameworks: tech.frameworks.append(fw) def _detect_team_indicators(self, message: str): """Detect if this is a team rather than individual""" team_indicators = [ r'\b(?:we|our|us)\b', r'\b(?:team|group|squad)\b', r'\b(?:my colleague|my coworker)\b', ] for pattern in team_indicators: if re.search(pattern, message, re.IGNORECASE): self.profile.is_team = True break def _update_work_patterns(self, timestamp: datetime): """Update work pattern information""" patterns = self.profile.work_patterns # Hour tracking hour = timestamp.hour patterns.active_hours[hour] = patterns.active_hours.get(hour, 0) + 1 # Day tracking day = timestamp.strftime('%A') patterns.active_days[day] = patterns.active_days.get(day, 0) + 1 def _extract_goals_and_interests(self, message: str): """Extract project goals and interests""" context = self.profile.project_context # Goal patterns goal_patterns = [ r"(?:want to|need to|trying to|going to)\s+([^.!?]+)", r"(?:goal is|objective is|aim is)\s+([^.!?]+)", ] for pattern in goal_patterns: match = re.search(pattern, message, re.IGNORECASE) if match: goal = match.group(1).strip() if goal not in context.current_goals: context.current_goals.append(goal) # Pain point patterns pain_patterns = [ r"(?:problem is|issue is|struggling with|stuck on)\s+([^.!?]+)", r"(?:doesn't work|not working|broken|failing)\s+([^.!?]+)", ] for pattern in pain_patterns: match = re.search(pattern, message, re.IGNORECASE) if match: pain = match.group(1).strip() if pain not in context.pain_points: context.pain_points.append(pain) def _get_detected_patterns(self, message: str) -> List[str]: """Get list of patterns detected in message""" patterns = [] if self._extract_name(message): patterns.append("name") if self._extract_pronouns(message): patterns.append("pronouns") if "?" in message: patterns.append("question") if any(word in message.lower() for word in ['thanks', 'thank you', 'appreciate']): patterns.append("gratitude") if any(word in message.lower() for word in ['please', 'could', 'would']): patterns.append("polite_request") return patterns def _log_learning_data(self, data: Dict[str, Any]): """Log learning data for analysis""" try: with open(self.learning_file, 'a') as f: f.write(json.dumps(data) + '\n') except Exception as e: print(f"Error logging learning data: {e}") def _synthesize_insights(self): """Synthesize insights from collected data""" # Generate summary summary_parts = [] if self.profile.primary_name: summary_parts.append(f"Primary name: {self.profile.primary_name}") if self.profile.is_team: summary_parts.append(f"Team of {self.profile.team_size} members") # Communication style insights style = self.profile.communication_style if style.formality_level > 0.7: summary_parts.append("Formal communication style") elif style.formality_level < 0.3: summary_parts.append("Casual communication style") if style.technical_depth > 0.7: summary_parts.append("Highly technical") # Technical profile insights tech = self.profile.technical_profile if tech.primary_languages: summary_parts.append(f"Works with: {', '.join(tech.primary_languages[:3])}") # Work patterns patterns = self.profile.work_patterns if patterns.active_hours: peak_hour = max(patterns.active_hours.items(), key=lambda x: x[1])[0] summary_parts.append(f"Most active around {peak_hour}:00") self.profile.summary = ". ".join(summary_parts) # Key traits traits = [] if style.formality_level > 0.7: traits.append("professional") if style.technical_depth > 0.7: traits.append("technical expert") if style.emoji_usage > 0.3: traits.append("expressive") if self.profile.engagement_level > 0.7: traits.append("highly engaged") self.profile.key_traits = traits def get_identity_summary(self) -> str: """Get a natural language summary of the steward's identity""" if not self.profile.primary_name and self.profile.total_interactions < 5: return "I haven't learned enough about you yet. Keep interacting with me and I'll build a better understanding of who you are." parts = [] # Name and identity if self.profile.primary_name: parts.append(f"You are {self.profile.primary_name}") if self.profile.aliases: parts[-1] += f" (also known as {', '.join(self.profile.aliases)})" # Team information if self.profile.is_team: parts.append(f"You're part of a team of {self.profile.team_size} people") # Technical profile tech = self.profile.technical_profile if tech.primary_languages: parts.append(f"You work with {', '.join(tech.primary_languages[:3])}") # Communication style style = self.profile.communication_style style_desc = [] if style.formality_level > 0.7: style_desc.append("formal") elif style.formality_level < 0.3: style_desc.append("casual") if style.technical_depth > 0.7: style_desc.append("technically sophisticated") if style_desc: parts.append(f"Your communication style is {' and '.join(style_desc)}") # Goals and context context = self.profile.project_context if context.current_goals: parts.append(f"You're working on: {', '.join(context.current_goals[:2])}") # Relationship if self.profile.relationship_quality > 0.7: parts.append("We have a great working relationship") return ". ".join(parts) + "." def get_essence_data(self) -> Dict[str, Any]: """Get essence data for the memory essence command""" return { "identity": { "name": self.profile.primary_name, "aliases": self.profile.aliases, "is_team": self.profile.is_team, "team_size": self.profile.team_size }, "personality": { "communication_style": { "formality": self.profile.communication_style.formality_level, "technical_depth": self.profile.communication_style.technical_depth, "emoji_usage": self.profile.communication_style.emoji_usage }, "key_traits": self.profile.key_traits }, "technical": { "languages": self.profile.technical_profile.primary_languages, "frameworks": self.profile.technical_profile.frameworks, "expertise": self.profile.technical_profile.expertise_areas }, "context": { "goals": self.profile.project_context.current_goals, "interests": self.profile.project_context.interests, "pain_points": self.profile.project_context.pain_points }, "relationship": { "quality": self.profile.relationship_quality, "trust": self.profile.trust_level, "engagement": self.profile.engagement_level, "total_interactions": self.profile.total_interactions }, "summary": self.profile.summary, "last_updated": self.profile.last_updated.isoformat() } # Global instance for easy access _profile_manager = None def get_steward_profile_manager() -> StewardProfileManager: """Get the global steward profile manager instance""" global _profile_manager if _profile_manager is None: _profile_manager = StewardProfileManager() return _profile_manager def learn_from_conversation(message: str, role: str = "user", timestamp: Optional[datetime] = None): """Convenience function to learn from a conversation message""" if role == "user": # Only learn from user messages manager = get_steward_profile_manager() manager.learn_from_message(message, timestamp) def get_steward_identity() -> str: """Get a summary of the steward's identity""" manager = get_steward_profile_manager() return manager.get_identity_summary() def get_steward_essence() -> Dict[str, Any]: """Get the essence data for the steward""" manager = get_steward_profile_manager() return manager.get_essence_data()