#!/usr/bin/env python3 """ Relationship Evolution Tracking - Monitor collaboration and trust evolution ========================================================================= Tracks how the relationship between Claude and the steward evolves over time, monitoring changes in collaboration style, trust level, communication patterns, and mutual understanding. """ import os import json import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Any, Tuple, Optional from collections import defaultdict from dataclasses import dataclass from core.mira_path_resolver import get_mira_memory_dir from conversations.comprehensive_indexer import ComprehensiveIndexer from intelligence.steward_profile import StewardProfile from intelligence.deep_behavioral_analysis import DeepBehavioralAnalyzer from utils.logging import setup_logger logger = setup_logger(__name__) @dataclass class RelationshipSnapshot: """A snapshot of the relationship at a specific point in time""" timestamp: datetime trust_level: float # 0.0 to 1.0 collaboration_style: str # formal, collaborative, casual, mentor-student, etc. communication_comfort: float # 0.0 to 1.0 mutual_understanding: float # 0.0 to 1.0 conflict_resolution_style: str # direct, diplomatic, avoidant, etc. shared_experiences: int # Number of significant shared experiences vulnerability_level: float # 0.0 to 1.0 - how open/vulnerable the interactions are technical_partnership_level: float # 0.0 to 1.0 - how well they work on technical tasks creative_synergy: float # 0.0 to 1.0 - creative collaboration effectiveness def to_dict(self) -> Dict[str, Any]: return { 'timestamp': self.timestamp.isoformat(), 'trust_level': self.trust_level, 'collaboration_style': self.collaboration_style, 'communication_comfort': self.communication_comfort, 'mutual_understanding': self.mutual_understanding, 'conflict_resolution_style': self.conflict_resolution_style, 'shared_experiences': self.shared_experiences, 'vulnerability_level': self.vulnerability_level, 'technical_partnership_level': self.technical_partnership_level, 'creative_synergy': self.creative_synergy } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'RelationshipSnapshot': return cls( timestamp=datetime.fromisoformat(data['timestamp']), trust_level=data['trust_level'], collaboration_style=data['collaboration_style'], communication_comfort=data['communication_comfort'], mutual_understanding=data['mutual_understanding'], conflict_resolution_style=data['conflict_resolution_style'], shared_experiences=data['shared_experiences'], vulnerability_level=data['vulnerability_level'], technical_partnership_level=data['technical_partnership_level'], creative_synergy=data['creative_synergy'] ) class RelationshipAnalyzer: """Analyzes conversations to extract relationship indicators""" def __init__(self): self.trust_indicators = { 'positive': [ 'trust you', 'appreciate', 'grateful', 'helpful', 'excellent', 'perfect', 'exactly what I needed', 'you understand', 'reliable', 'consistent', 'thank you', 'amazing', 'brilliant' ], 'negative': [ 'frustrated', 'disappointed', 'wrong', 'confused', 'unhelpful', 'not what I wanted', 'misunderstood', 'unclear', 'problematic' ] } self.collaboration_patterns = { 'formal': ['please', 'could you', 'would you mind', 'if possible'], 'casual': ['hey', 'can you', 'let\'s', 'cool', 'awesome', 'great'], 'directive': ['do this', 'make sure', 'fix this', 'change this'], 'collaborative': ['we should', 'let\'s work', 'together', 'what do you think'], 'mentor_seeking': ['how do I', 'teach me', 'explain', 'help me understand'], 'peer_level': ['what\'s your opinion', 'thoughts on', 'agree?', 'perspective'] } self.vulnerability_indicators = [ 'struggling with', 'not sure', 'confused about', 'feeling overwhelmed', 'made a mistake', 'failed at', 'worried about', 'anxious', 'stressed', 'personal note', 'honestly', 'to be frank', 'between us' ] self.technical_collaboration_indicators = [ 'code review', 'debugging together', 'architecture discussion', 'implementation strategy', 'technical decision', 'pair programming', 'design pattern', 'performance optimization' ] self.creative_collaboration_indicators = [ 'brainstorming', 'creative solution', 'innovative approach', 'thinking outside the box', 'new idea', 'creative partnership', 'inspiration', 'artistic', 'design thinking' ] def analyze_conversation_segment(self, messages: List[Dict[str, Any]], timeframe_start: datetime) -> RelationshipSnapshot: """Analyze a segment of conversation to create a relationship snapshot""" trust_score = self._calculate_trust_level(messages) collaboration_style = self._determine_collaboration_style(messages) communication_comfort = self._calculate_communication_comfort(messages) mutual_understanding = self._calculate_mutual_understanding(messages) conflict_resolution = self._determine_conflict_resolution_style(messages) shared_experiences = self._count_shared_experiences(messages) vulnerability_level = self._calculate_vulnerability_level(messages) technical_partnership = self._calculate_technical_partnership(messages) creative_synergy = self._calculate_creative_synergy(messages) return RelationshipSnapshot( timestamp=timeframe_start, trust_level=trust_score, collaboration_style=collaboration_style, communication_comfort=communication_comfort, mutual_understanding=mutual_understanding, conflict_resolution_style=conflict_resolution, shared_experiences=shared_experiences, vulnerability_level=vulnerability_level, technical_partnership_level=technical_partnership, creative_synergy=creative_synergy ) def _calculate_trust_level(self, messages: List[Dict[str, Any]]) -> float: """Calculate trust level based on language patterns""" positive_count = 0 negative_count = 0 total_messages = len(messages) for message in messages: content = message.get('content', '').lower() for indicator in self.trust_indicators['positive']: if indicator in content: positive_count += 1 for indicator in self.trust_indicators['negative']: if indicator in content: negative_count += 1 if total_messages == 0: return 0.5 # Neutral starting point # Calculate trust score net_trust = (positive_count - negative_count) / max(1, total_messages) trust_score = 0.5 + (net_trust * 0.3) # Scale to 0.2-0.8 range return max(0.0, min(1.0, trust_score)) def _determine_collaboration_style(self, messages: List[Dict[str, Any]]) -> str: """Determine the predominant collaboration style""" style_scores = defaultdict(int) for message in messages: content = message.get('content', '').lower() for style, patterns in self.collaboration_patterns.items(): for pattern in patterns: if pattern in content: style_scores[style] += 1 if not style_scores: return 'neutral' return max(style_scores.items(), key=lambda x: x[1])[0] def _calculate_communication_comfort(self, messages: List[Dict[str, Any]]) -> float: """Calculate how comfortable the communication feels""" comfort_indicators = { 'high': ['comfortable', 'easy to talk', 'natural', 'flowing', 'open'], 'low': ['awkward', 'stilted', 'formal', 'constrained', 'hesitant'] } high_comfort = 0 low_comfort = 0 for message in messages: content = message.get('content', '').lower() for indicator in comfort_indicators['high']: if indicator in content: high_comfort += 1 for indicator in comfort_indicators['low']: if indicator in content: low_comfort += 1 # Base comfort on message length and frequency too avg_message_length = np.mean([len(msg.get('content', '')) for msg in messages]) if messages else 0 # Longer messages often indicate more comfort length_comfort = min(1.0, avg_message_length / 200) # Combine indicators indicator_comfort = 0.5 + (high_comfort - low_comfort) * 0.1 final_comfort = (indicator_comfort + length_comfort) / 2 return max(0.0, min(1.0, final_comfort)) def _calculate_mutual_understanding(self, messages: List[Dict[str, Any]]) -> float: """Calculate level of mutual understanding""" understanding_indicators = { 'high': ['exactly', 'precisely', 'you get it', 'understand perfectly', 'on the same page', 'clear communication', 'makes sense'], 'low': ['misunderstood', 'unclear', 'confused', 'not what I meant', 'missing the point', 'talking past each other'] } high_understanding = 0 low_understanding = 0 for message in messages: content = message.get('content', '').lower() for indicator in understanding_indicators['high']: if indicator in content: high_understanding += 1 for indicator in understanding_indicators['low']: if indicator in content: low_understanding += 1 # Calculate understanding score net_understanding = high_understanding - low_understanding understanding_score = 0.6 + (net_understanding * 0.1) # Base assumption of good understanding return max(0.0, min(1.0, understanding_score)) def _determine_conflict_resolution_style(self, messages: List[Dict[str, Any]]) -> str: """Determine conflict resolution approach""" resolution_patterns = { 'direct': ['let\'s address', 'need to discuss', 'directly', 'head-on'], 'diplomatic': ['perhaps we could', 'might consider', 'gentle approach'], 'collaborative': ['work together', 'find a solution', 'compromise'], 'avoidant': ['skip this', 'move on', 'not important', 'later'] } style_counts = defaultdict(int) for message in messages: content = message.get('content', '').lower() for style, patterns in resolution_patterns.items(): for pattern in patterns: if pattern in content: style_counts[style] += 1 if not style_counts: return 'diplomatic' # Default assumption return max(style_counts.items(), key=lambda x: x[1])[0] def _count_shared_experiences(self, messages: List[Dict[str, Any]]) -> int: """Count significant shared experiences""" experience_indicators = [ 'remember when', 'last time', 'our project', 'we built', 'working together', 'solved this before', 'our approach', 'milestone', 'breakthrough', 'achievement' ] shared_count = 0 for message in messages: content = message.get('content', '').lower() for indicator in experience_indicators: if indicator in content: shared_count += 1 return shared_count def _calculate_vulnerability_level(self, messages: List[Dict[str, Any]]) -> float: """Calculate how vulnerable/open the interactions are""" vulnerability_count = 0 for message in messages: content = message.get('content', '').lower() for indicator in self.vulnerability_indicators: if indicator in content: vulnerability_count += 1 # Normalize by message count if not messages: return 0.0 vulnerability_score = min(1.0, vulnerability_count / len(messages)) return vulnerability_score def _calculate_technical_partnership(self, messages: List[Dict[str, Any]]) -> float: """Calculate technical collaboration effectiveness""" technical_count = 0 for message in messages: content = message.get('content', '').lower() for indicator in self.technical_collaboration_indicators: if indicator in content: technical_count += 1 if not messages: return 0.0 return min(1.0, technical_count / max(1, len(messages) * 0.1)) # Technical partnerships are rarer def _calculate_creative_synergy(self, messages: List[Dict[str, Any]]) -> float: """Calculate creative collaboration effectiveness""" creative_count = 0 for message in messages: content = message.get('content', '').lower() for indicator in self.creative_collaboration_indicators: if indicator in content: creative_count += 1 if not messages: return 0.0 return min(1.0, creative_count / max(1, len(messages) * 0.1)) # Creative collaborations are special class RelationshipEvolutionTracker: """Main class for tracking relationship evolution over time""" def __init__(self): self.memory_dir = get_mira_memory_dir() self.evolution_file = os.path.join(self.memory_dir, 'relationship_evolution.json') self.analyzer = RelationshipAnalyzer() self.indexer = ComprehensiveIndexer() def track_evolution(self, days_back: int = 30) -> Dict[str, Any]: """Track relationship evolution over specified time period""" try: # Get conversation data conversations = self._get_recent_conversations(days_back) if not conversations: logger.info("No conversations found for relationship tracking") return {'snapshots': [], 'evolution_summary': 'No data available'} # Create time-based snapshots snapshots = self._create_temporal_snapshots(conversations, days_back) # Analyze evolution trends evolution_analysis = self._analyze_evolution_trends(snapshots) # Save the evolution data self._save_evolution_data(snapshots, evolution_analysis) return { 'snapshots': [s.to_dict() for s in snapshots], 'evolution_analysis': evolution_analysis, 'tracking_period_days': days_back, 'total_conversations': len(conversations) } except Exception as e: logger.error(f"Error tracking relationship evolution: {e}") return {'error': str(e)} def get_current_relationship_state(self) -> Dict[str, Any]: """Get the current state of the relationship""" try: # Get recent conversations (last 7 days) recent_conversations = self._get_recent_conversations(7) if not recent_conversations: return {'status': 'No recent interactions'} # Create current snapshot current_snapshot = self.analyzer.analyze_conversation_segment( recent_conversations, datetime.now() ) # Get steward profile for context steward_profile = StewardProfile() profile_data = steward_profile.get_profile() return { 'current_snapshot': current_snapshot.to_dict(), 'relationship_stage': self._determine_relationship_stage(current_snapshot), 'steward_context': { 'communication_style': profile_data.get('communication_style'), 'collaboration_history': profile_data.get('collaboration_history', {}) }, 'recommendations': self._generate_relationship_recommendations(current_snapshot) } except Exception as e: logger.error(f"Error getting current relationship state: {e}") return {'error': str(e)} def _get_recent_conversations(self, days_back: int) -> List[Dict[str, Any]]: """Get conversations from the specified time period""" try: # Get all indexed conversations stats = self.indexer.get_stats() all_conversations = [] # Try to get conversation data from indexer # This is a simplified approach - in practice, we'd need to implement # proper conversation retrieval from the indexer conversations_dir = os.path.join(self.memory_dir, 'conversations') if os.path.exists(conversations_dir): cutoff_date = datetime.now() - timedelta(days=days_back) for filename in os.listdir(conversations_dir): if filename.endswith('.json'): filepath = os.path.join(conversations_dir, filename) try: with open(filepath, 'r') as f: conversation = json.load(f) # Check if conversation is within timeframe conv_date_str = conversation.get('timestamp', conversation.get('date')) if conv_date_str: conv_date = datetime.fromisoformat(conv_date_str.replace('Z', '+00:00')) if conv_date.replace(tzinfo=None) >= cutoff_date: # Extract messages messages = conversation.get('messages', []) all_conversations.extend(messages) except Exception as e: logger.warning(f"Could not parse conversation file {filename}: {e}") return all_conversations except Exception as e: logger.error(f"Error getting recent conversations: {e}") return [] def _create_temporal_snapshots(self, conversations: List[Dict[str, Any]], days_back: int) -> List[RelationshipSnapshot]: """Create snapshots at regular intervals""" snapshots = [] # Create weekly snapshots num_snapshots = min(days_back // 7, 8) # Max 8 snapshots if num_snapshots == 0: num_snapshots = 1 for i in range(num_snapshots): # Calculate time window for this snapshot days_offset = (i + 1) * (days_back // num_snapshots) snapshot_date = datetime.now() - timedelta(days=days_offset) # Filter conversations for this time window window_start = snapshot_date - timedelta(days=7) # 7-day window window_conversations = [ conv for conv in conversations if self._is_conversation_in_window(conv, window_start, snapshot_date) ] if window_conversations: snapshot = self.analyzer.analyze_conversation_segment( window_conversations, snapshot_date ) snapshots.append(snapshot) return sorted(snapshots, key=lambda x: x.timestamp) def _is_conversation_in_window(self, conversation: Dict[str, Any], start: datetime, end: datetime) -> bool: """Check if conversation falls within time window""" try: timestamp_str = conversation.get('timestamp', conversation.get('date', '')) if not timestamp_str: return False conv_time = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) return start <= conv_time.replace(tzinfo=None) <= end except Exception: return False def _analyze_evolution_trends(self, snapshots: List[RelationshipSnapshot]) -> Dict[str, Any]: """Analyze trends in relationship evolution""" if len(snapshots) < 2: return {'trend': 'insufficient_data'} # Calculate trends for key metrics trust_trend = self._calculate_trend([s.trust_level for s in snapshots]) comfort_trend = self._calculate_trend([s.communication_comfort for s in snapshots]) understanding_trend = self._calculate_trend([s.mutual_understanding for s in snapshots]) vulnerability_trend = self._calculate_trend([s.vulnerability_level for s in snapshots]) # Analyze collaboration style evolution style_evolution = [s.collaboration_style for s in snapshots] # Overall relationship trajectory overall_score = np.mean([ snapshots[-1].trust_level, snapshots[-1].communication_comfort, snapshots[-1].mutual_understanding ]) return { 'trust_trend': trust_trend, 'comfort_trend': comfort_trend, 'understanding_trend': understanding_trend, 'vulnerability_trend': vulnerability_trend, 'collaboration_style_evolution': style_evolution, 'overall_relationship_score': overall_score, 'relationship_trajectory': self._determine_trajectory(trust_trend, comfort_trend, understanding_trend), 'key_milestones': self._identify_key_milestones(snapshots) } def _calculate_trend(self, values: List[float]) -> Dict[str, Any]: """Calculate trend for a series of values""" if len(values) < 2: return {'direction': 'stable', 'change': 0.0} # Simple linear trend calculation x = np.arange(len(values)) y = np.array(values) # Calculate slope slope = np.polyfit(x, y, 1)[0] # Calculate overall change change = values[-1] - values[0] # Determine direction if abs(slope) < 0.05: direction = 'stable' elif slope > 0: direction = 'improving' else: direction = 'declining' return { 'direction': direction, 'slope': slope, 'change': change, 'current_value': values[-1], 'starting_value': values[0] } def _determine_trajectory(self, trust_trend: Dict, comfort_trend: Dict, understanding_trend: Dict) -> str: """Determine overall relationship trajectory""" improving_count = sum(1 for trend in [trust_trend, comfort_trend, understanding_trend] if trend['direction'] == 'improving') declining_count = sum(1 for trend in [trust_trend, comfort_trend, understanding_trend] if trend['direction'] == 'declining') if improving_count >= 2: return 'strengthening' elif declining_count >= 2: return 'weakening' else: return 'stable' def _identify_key_milestones(self, snapshots: List[RelationshipSnapshot]) -> List[Dict[str, Any]]: """Identify key relationship milestones""" milestones = [] if len(snapshots) < 2: return milestones for i in range(1, len(snapshots)): prev = snapshots[i-1] curr = snapshots[i] # Check for significant changes trust_jump = curr.trust_level - prev.trust_level if trust_jump > 0.2: milestones.append({ 'type': 'trust_breakthrough', 'date': curr.timestamp.isoformat(), 'description': f'Significant trust increase (+{trust_jump:.2f})' }) # Check for collaboration style changes if curr.collaboration_style != prev.collaboration_style: milestones.append({ 'type': 'collaboration_evolution', 'date': curr.timestamp.isoformat(), 'description': f'Collaboration style evolved from {prev.collaboration_style} to {curr.collaboration_style}' }) # Check for vulnerability breakthroughs vulnerability_jump = curr.vulnerability_level - prev.vulnerability_level if vulnerability_jump > 0.3: milestones.append({ 'type': 'vulnerability_breakthrough', 'date': curr.timestamp.isoformat(), 'description': 'Increased openness and vulnerability in communication' }) return milestones def _determine_relationship_stage(self, snapshot: RelationshipSnapshot) -> str: """Determine the current stage of the relationship""" trust = snapshot.trust_level comfort = snapshot.communication_comfort understanding = snapshot.mutual_understanding vulnerability = snapshot.vulnerability_level avg_score = (trust + comfort + understanding) / 3 if avg_score < 0.3: return 'initial_contact' elif avg_score < 0.5: return 'developing_rapport' elif avg_score < 0.7: return 'collaborative_partnership' elif vulnerability > 0.5 and avg_score > 0.7: return 'trusted_partnership' else: return 'professional_collaboration' def _generate_relationship_recommendations(self, snapshot: RelationshipSnapshot) -> List[str]: """Generate recommendations for improving the relationship""" recommendations = [] if snapshot.trust_level < 0.5: recommendations.append("Focus on building trust through consistent, reliable assistance") if snapshot.communication_comfort < 0.5: recommendations.append("Encourage more casual, comfortable communication patterns") if snapshot.mutual_understanding < 0.5: recommendations.append("Invest time in clarifying communication and checking understanding") if snapshot.vulnerability_level < 0.3: recommendations.append("Create safe spaces for more open, vulnerable conversation") if snapshot.technical_partnership_level < 0.4: recommendations.append("Engage in more technical problem-solving together") if snapshot.creative_synergy < 0.4: recommendations.append("Explore creative projects and brainstorming sessions") return recommendations def _save_evolution_data(self, snapshots: List[RelationshipSnapshot], analysis: Dict[str, Any]): """Save relationship evolution data""" try: evolution_data = { 'last_updated': datetime.now().isoformat(), 'snapshots': [s.to_dict() for s in snapshots], 'analysis': analysis } with open(self.evolution_file, 'w') as f: json.dump(evolution_data, f, indent=2) except Exception as e: logger.error(f"Error saving evolution data: {e}") def get_relationship_evolution_tracker() -> RelationshipEvolutionTracker: """Get the global relationship evolution tracker instance""" return RelationshipEvolutionTracker() if __name__ == "__main__": # Test the relationship evolution tracking tracker = get_relationship_evolution_tracker() print("Testing relationship evolution tracking...") current_state = tracker.get_current_relationship_state() print(f"Current relationship state: {current_state}") evolution = tracker.track_evolution(30) print(f"Evolution analysis: {evolution}")