Coverage for core / platform_paths.py: 55.0%

1""" 

2Cross-platform data directory resolution for Nunba / HARTOS. 

3 

4Returns the correct data root for each platform: 

5 Windows: ~/Documents/Nunba 

6 macOS: ~/Library/Application Support/Nunba 

7 Linux: ~/.config/nunba 

8 HARTOS OS (embedded): /var/lib/hartos (or HARTOS_DATA_DIR env var) 

9 

10Override with NUNBA_DATA_DIR env var for any custom deployment. 

11 

12Usage: 

13 from core.platform_paths import get_data_dir, get_db_path, get_agent_data_dir 

14 data_root = get_data_dir() # e.g. ~/Documents/Nunba on Windows 

15 db_path = get_db_path() # .../data/hevolve_database.db 

16 agent_dir = get_agent_data_dir() # .../data/agent_data 

17""" 

18 

19import os 

20import sys 

21import time 

22 

23_IS_WINDOWS = sys.platform == 'win32' 

24_IS_MACOS = sys.platform == 'darwin' 

25_IS_LINUX = sys.platform.startswith('linux') 

26 

27_cached_data_dir = None 

28 

29 

30def get_data_dir() -> str: 

31 """Return the platform-appropriate Nunba data root directory. 

32 

33 Priority: 

34 1. NUNBA_DATA_DIR env var (explicit override) 

35 2. HARTOS_DATA_DIR env var (embedded OS / custom deployment) 

36 3. Platform default 

37 """ 

38 global _cached_data_dir 

39 if _cached_data_dir is not None: 

40 return _cached_data_dir 

41 

42 # 1. Explicit override 

43 override = os.environ.get('NUNBA_DATA_DIR', '').strip() 

44 if override: 

45 _cached_data_dir = override 

46 return _cached_data_dir 

47 

48 # 2. HARTOS OS deployment override 

49 hartos_dir = os.environ.get('HARTOS_DATA_DIR', '').strip() 

50 if hartos_dir: 

51 _cached_data_dir = hartos_dir 

52 return _cached_data_dir 

53 

54 # 3. Detect embedded HARTOS OS (systemd service, no home dir) 

55 if _IS_LINUX and os.path.isfile('/etc/hartos-release'): 

56 _cached_data_dir = '/var/lib/hartos' 

57 return _cached_data_dir 

58 

59 # 4. Platform defaults 

60 home = os.path.expanduser('~') 

61 if _IS_WINDOWS: 

62 _cached_data_dir = os.path.join(home, 'Documents', 'Nunba') 

63 elif _IS_MACOS: 

64 _cached_data_dir = os.path.join(home, 'Library', 'Application Support', 'Nunba') 

65 else: 

66 # Linux / other Unix 

67 xdg = os.environ.get('XDG_DATA_HOME', '').strip() 

68 if xdg: 

69 _cached_data_dir = os.path.join(xdg, 'nunba') 

70 else: 

71 _cached_data_dir = os.path.join(home, '.config', 'nunba') 

72 

73 return _cached_data_dir 

74 

75 

76def get_db_dir() -> str: 

77 """Return the data/ subdirectory (databases, caches).""" 

78 return os.path.join(get_data_dir(), 'data') 

79 

80 

81def get_db_path(filename: str = 'hevolve_database.db') -> str: 

82 """Return full path to a database file inside data/.""" 

83 return os.path.join(get_db_dir(), filename) 

84 

85 

86def get_agent_data_dir() -> str: 

87 """Return the agent_data/ subdirectory.""" 

88 return os.path.join(get_db_dir(), 'agent_data') 

89 

90 

91def get_coding_workspace_dir() -> str: 

92 """Return the AutoGen UserProxyAgent code-execution workspace. 

93 

94 Routes off CWD to a writable user-data path so install-tree CWD 

95 (e.g. C:\\Program Files (x86)\\HevolveAI\\Nunba) doesn't fail 

96 `[WinError 5] Access is denied: 'coding'`. 

97 """ 

98 path = os.path.join(get_db_dir(), 'coding') 

99 os.makedirs(path, exist_ok=True) 

100 return path 

101 

102 

103def get_prompts_dir() -> str: 

104 """Return the prompts/ subdirectory.""" 

105 return os.path.join(get_db_dir(), 'prompts') 

106 

107 

108def get_log_dir() -> str: 

109 """Return the platform-appropriate log directory.""" 

110 if _IS_WINDOWS: 

111 return os.path.join(get_data_dir(), 'logs') 

112 elif _IS_MACOS: 

113 return os.path.expanduser('~/Library/Logs/Nunba') 

114 else: 

115 return os.path.join(get_data_dir(), 'logs') 

116 

117 

118def get_memory_graph_dir(session_key: str = '') -> str: 

119 """Return the memory_graph/ subdirectory, optionally with session key.""" 

120 base = os.path.join(get_db_dir(), 'memory_graph') 

121 if session_key: 

122 return os.path.join(base, session_key) 

123 return base 

124 

125 

126def get_simplemem_dir(session_key: str = '') -> str: 

127 """Return the simplemem_db/ subdirectory, optionally with session key.""" 

128 base = os.path.join(get_db_dir(), 'simplemem_db') 

129 if session_key: 

130 return os.path.join(base, session_key) 

131 return base 

132 

133 

134def cleanup_old_logs(max_age_days: int = 7, max_total_mb: int = 50): 

135 """Delete log files older than max_age_days or when total exceeds max_total_mb. 

136 

137 Called at startup to prevent unbounded log accumulation. 

138 Safe: only deletes *.log and *.log.* files in the log directory. 

139 """ 

140 import glob as _glob 

141 log_dir = get_log_dir() 

142 if not os.path.isdir(log_dir): 

143 return 

144 now = time.time() 

145 cutoff = now - (max_age_days * 86400) 

146 log_patterns = [ 

147 os.path.join(log_dir, '*.log'), 

148 os.path.join(log_dir, '*.log.*'), 

149 ] 

150 all_logs = [] 

151 for pat in log_patterns: 

152 all_logs.extend(_glob.glob(pat)) 

153 # Sort oldest first 

154 all_logs.sort(key=lambda f: os.path.getmtime(f) if os.path.exists(f) else 0) 

155 

156 deleted = 0 

157 # Phase 1: delete files older than max_age_days 

158 for f in all_logs: 

159 try: 

160 if os.path.getmtime(f) < cutoff: 

161 os.remove(f) 

162 deleted += 1 

163 except OSError: 

164 pass 

165 

166 # Phase 2: if still over budget, delete oldest until under limit 

167 remaining = [f for f in all_logs if os.path.exists(f)] 

168 total_bytes = sum(os.path.getsize(f) for f in remaining if os.path.exists(f)) 

169 max_bytes = max_total_mb * 1024 * 1024 

170 for f in remaining: 

171 if total_bytes <= max_bytes: 

172 break 

173 try: 

174 sz = os.path.getsize(f) 

175 os.remove(f) 

176 total_bytes -= sz 

177 deleted += 1 

178 except OSError: 

179 pass 

180 

181 if deleted: 

182 import logging 

183 logging.getLogger('hevolve.platform').info( 

184 f"Log cleanup: deleted {deleted} old log files from {log_dir}") 

185 

186 

187def ensure_data_dirs(): 

188 """Create all standard data directories if they don't exist. 

189 

190 Also runs log cleanup on startup to prevent unbounded log accumulation. 

191 """ 

192 for d in [get_db_dir(), get_agent_data_dir(), get_prompts_dir(), 

193 get_log_dir(), get_memory_graph_dir(), get_simplemem_dir()]: 

194 os.makedirs(d, exist_ok=True) 

195 # Clean old logs on every startup (safe — worst case is a no-op) 

196 try: 

197 cleanup_old_logs(max_age_days=7, max_total_mb=50) 

198 except Exception: 

199 pass 

200 

201 

202def reset_cache(): 

203 """Reset the cached data dir (useful for testing).""" 

204 global _cached_data_dir 

205 _cached_data_dir = None