Coverage for core / resonance_profile.py: 98.5%

68 statements  

« prev     ^ index     » next       coverage.py v7.14.0, created at 2026-05-12 04:49 +0000

1""" 

2UserResonanceProfile — Per-user continuous personality tuning parameters. 

3 

4Each user_id gets a learned profile with continuous floats (0.0-1.0) 

5instead of binary switches. Stored as JSON at: 

6 agent_data/resonance/{user_id}_resonance.json 

7 

8Reuses agent_data/ storage pattern from helper_ledger.py (DRY). 

9""" 

10 

11import json 

12import logging 

13import os 

14import time 

15from dataclasses import dataclass, field, asdict 

16from typing import Dict, List, Optional, Any 

17 

18logger = logging.getLogger(__name__) 

19 

20# Default tuning parameters — neutral midpoint for all dimensions 

21DEFAULT_TUNING = { 

22 'formality_score': 0.5, # 0.0=very casual, 1.0=very formal 

23 'verbosity_score': 0.5, # 0.0=terse, 1.0=very detailed 

24 'warmth_score': 0.6, # 0.0=professional distance, 1.0=very warm 

25 'pace_score': 0.5, # 0.0=slow/thorough, 1.0=fast/action-oriented 

26 'technical_depth': 0.5, # 0.0=simple, 1.0=highly technical 

27 'encouragement_level': 0.6, # 0.0=matter-of-fact, 1.0=very encouraging 

28 'humor_receptivity': 0.3, # 0.0=serious, 1.0=playful 

29 'autonomy_preference': 0.5, # 0.0=ask before acting, 1.0=act autonomously 

30} 

31 

32# Ordered dimension keys — index-stable for matrix operations 

33TUNING_DIM_KEYS = list(DEFAULT_TUNING.keys()) 

34TUNING_DIM_COUNT = len(TUNING_DIM_KEYS) # 8 

35 

36RESONANCE_STORAGE_DIR = os.environ.get( 

37 'RESONANCE_STORAGE_DIR', 

38 os.path.join('agent_data', 'resonance') 

39) 

40 

41 

42@dataclass 

43class UserResonanceProfile: 

44 """Per-user continuous tuning profile — the 'frequency' for this user.""" 

45 

46 user_id: str = "" 

47 

48 # Multi-dimensional tuning (continuous floats 0.0-1.0) 

49 tuning: Dict[str, float] = field(default_factory=lambda: dict(DEFAULT_TUNING)) 

50 

51 # Biometric signatures (optional, stored as list of floats) 

52 face_embedding: Optional[List[float]] = None 

53 voice_embedding: Optional[List[float]] = None 

54 face_enrollment_count: int = 0 

55 voice_enrollment_count: int = 0 

56 

57 # Interaction patterns learned 

58 avg_message_length: float = 0.0 

59 avg_response_time_ms: float = 0.0 

60 vocabulary_complexity: float = 0.5 # 0=simple, 1=complex 

61 topic_preferences: Dict[str, float] = field(default_factory=dict) 

62 

63 # Oscillation tracking (HARTOS detects, HevolveAI corrects) 

64 tuning_history: List[List[float]] = field(default_factory=list) # last 20 snapshots 

65 gradient_active: bool = False # True when oscillation detected, flags HevolveAI 

66 

67 # Per-user EMA alpha (None = use global default from RESONANCE_EMA_ALPHA) 

68 ema_alpha: Optional[float] = None 

69 

70 # Metadata 

71 total_interactions: int = 0 

72 resonance_confidence: float = 0.0 # 0.0=untuned, 1.0=highly tuned 

73 last_interaction_at: float = 0.0 

74 created_at: float = field(default_factory=time.time) 

75 updated_at: float = field(default_factory=time.time) 

76 

77 def to_dict(self) -> Dict[str, Any]: 

78 return asdict(self) 

79 

80 @classmethod 

81 def from_dict(cls, data: Dict[str, Any]) -> 'UserResonanceProfile': 

82 """Deserialize, merging any missing tuning keys from defaults.""" 

83 if 'tuning' in data: 

84 for key, default in DEFAULT_TUNING.items(): 

85 data['tuning'].setdefault(key, default) 

86 known = {k for k in cls.__dataclass_fields__} 

87 return cls(**{k: v for k, v in data.items() if k in known}) 

88 

89 def get_tuning(self, key: str) -> float: 

90 return self.tuning.get(key, DEFAULT_TUNING.get(key, 0.5)) 

91 

92 def set_tuning(self, key: str, value: float) -> None: 

93 self.tuning[key] = max(0.0, min(1.0, value)) 

94 self.updated_at = time.time() 

95 

96 

97# ═══════════════════════════════════════════════════════════════════════ 

98# Persistence 

99# ═══════════════════════════════════════════════════════════════════════ 

100 

101def save_resonance_profile(profile: UserResonanceProfile, 

102 base_dir: str = None) -> None: 

103 """Save profile to agent_data/resonance/{user_id}_resonance.json. 

104 

105 Encrypted at rest when HEVOLVE_DATA_KEY is configured. 

106 Falls back to plaintext JSON when encryption key is not set. 

107 """ 

108 base_dir = base_dir or RESONANCE_STORAGE_DIR 

109 os.makedirs(base_dir, exist_ok=True) 

110 path = os.path.join(base_dir, f"{profile.user_id}_resonance.json") 

111 try: 

112 from security.crypto import encrypt_json_file 

113 encrypt_json_file(path, profile.to_dict()) 

114 except ImportError: 

115 # Fallback: no crypto module available 

116 with open(path, 'w') as f: 

117 json.dump(profile.to_dict(), f, indent=2) 

118 except Exception as e: 

119 logger.warning(f"Failed to save resonance profile: {e}") 

120 

121 

122def load_resonance_profile(user_id: str, 

123 base_dir: str = None) -> Optional[UserResonanceProfile]: 

124 """Load profile from disk. Auto-detects encrypted vs plaintext. 

125 

126 Returns None if not found. 

127 """ 

128 base_dir = base_dir or RESONANCE_STORAGE_DIR 

129 path = os.path.join(base_dir, f"{user_id}_resonance.json") 

130 if not os.path.exists(path): 

131 return None 

132 try: 

133 from security.crypto import decrypt_json_file 

134 data = decrypt_json_file(path) 

135 if data is None: 

136 return None 

137 return UserResonanceProfile.from_dict(data) 

138 except ImportError: 

139 # Fallback: no crypto module 

140 with open(path, 'r') as f: 

141 data = json.load(f) 

142 return UserResonanceProfile.from_dict(data) 

143 except Exception as e: 

144 logger.warning(f"Failed to load resonance profile: {e}") 

145 return None 

146 

147 

148def get_or_create_profile(user_id: str, 

149 base_dir: str = None) -> UserResonanceProfile: 

150 """Load existing profile or create a fresh one.""" 

151 profile = load_resonance_profile(user_id, base_dir) 

152 if profile is None: 

153 profile = UserResonanceProfile(user_id=user_id) 

154 return profile