Coverage for src / infra / tools / env.py: 57%

1"""Environment configuration and loading for mala. 

2 

3Centralizes config paths and dotenv loading. Import this module early 

4to ensure environment variables are set before Braintrust setup. 

5""" 

6 

7import os 

8from pathlib import Path 

9 

10from dotenv import load_dotenv 

11 

12 

13# User config directory (stores .env, runs, etc.) 

14# Respects XDG_CONFIG_HOME if set, otherwise defaults to ~/.config 

15def _get_xdg_config_home() -> Path: 

16 """Get XDG_CONFIG_HOME or default to ~/.config.""" 

17 return Path(os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))) 

18 

19 

20USER_CONFIG_DIR = _get_xdg_config_home() / "mala" 

21 

22 

23def get_cache_dir() -> Path: 

24 """Get the mala cache directory. 

25 

26 Returns the path to mala's cache directory, following XDG conventions. 

27 Uses $XDG_CONFIG_HOME/mala/.cache if XDG_CONFIG_HOME is set, 

28 otherwise ~/.config/mala/.cache. 

29 

30 The directory is created if it doesn't exist. 

31 

32 Returns: 

33 Path to the cache directory. 

34 """ 

35 cache_dir = USER_CONFIG_DIR / ".cache" 

36 cache_dir.mkdir(parents=True, exist_ok=True) 

37 return cache_dir 

38 

39 

40# Run metadata directory 

41# Can be overridden via MALA_RUNS_DIR environment variable 

42def get_runs_dir() -> Path: 

43 """Get the runs directory, respecting MALA_RUNS_DIR env var. 

44 

45 This function evaluates the env var at call time, so it respects 

46 values loaded from .env via load_user_env(). 

47 """ 

48 return Path(os.environ.get("MALA_RUNS_DIR", str(USER_CONFIG_DIR / "runs"))) 

49 

50 

51def get_repo_runs_dir(repo_path: Path) -> Path: 

52 """Get runs directory segmented by repo path. 

53 

54 Converts repo path to a safe directory name by replacing '/' with '-'. 

55 Example: /home/cyou/mala -> -home-cyou-mala 

56 

57 Args: 

58 repo_path: Repository path to segment runs by. 

59 

60 Returns: 

61 Path to the repo-specific runs directory. 

62 """ 

63 safe_name = encode_repo_path(repo_path) 

64 return get_runs_dir() / safe_name 

65 

66 

67# Lock directory for multi-agent coordination 

68# Can be overridden via MALA_LOCK_DIR environment variable 

69def get_lock_dir() -> Path: 

70 """Get the lock directory, respecting MALA_LOCK_DIR env var. 

71 

72 This function evaluates the env var at call time, so it respects 

73 values loaded from .env via load_user_env(). 

74 """ 

75 return Path(os.environ.get("MALA_LOCK_DIR", "/tmp/mala-locks")) 

76 

77 

78# Lock scripts directory (relative to this file: src/infra/tools/env.py -> src/scripts/) 

79SCRIPTS_DIR = Path(__file__).resolve().parents[2] / "scripts" 

80 

81# Prompts directory (relative to this file: src/infra/tools/env.py -> src/prompts/) 

82PROMPTS_DIR = Path(__file__).resolve().parents[2] / "prompts" 

83 

84 

85def load_user_env() -> None: 

86 """Load environment from user config directory. 

87 

88 Loads ${USER_CONFIG_DIR}/.env (typically ~/.config/mala/.env). 

89 Call this early for Braintrust API key setup before SDK imports. 

90 """ 

91 load_dotenv(dotenv_path=USER_CONFIG_DIR / ".env") 

92 

93 

94def load_env(repo_path: Path | None = None) -> None: 

95 """Load environment from user config and optionally repo. 

96 

97 NOTE: The repo_path parameter is for TESTING ONLY. Production code should 

98 only use load_user_env() to load from ~/.config/mala/.env. 

99 

100 Args: 

101 repo_path: Optional repository path. If provided, loads <repo_path>/.env 

102 with override=True. FOR TESTING ONLY. 

103 """ 

104 load_user_env() 

105 if repo_path is not None: 

106 load_dotenv(dotenv_path=repo_path / ".env", override=True) 

107 

108 

109def encode_repo_path(repo_path: Path) -> str: 

110 """Encode repo path to match Claude SDK project directory naming. 

111 

112 Claude SDK stores session logs in ~/.claude/projects/{encoded-path}/. 

113 The encoding replaces path separators with hyphens, normalizes underscores 

114 to hyphens, and prefixes with a hyphen. 

115 

116 Example: /home/cyou/mala -> -home-cyou-mala 

117 

118 Args: 

119 repo_path: Repository path to encode. 

120 

121 Returns: 

122 Encoded path string suitable for Claude projects directory. 

123 """ 

124 resolved = repo_path.resolve() 

125 # Skip root and join parts with hyphens, prefix with hyphen. 

126 # Normalize underscores to hyphens to match Claude SDK project dir naming. 

127 encoded = "-" + "-".join(resolved.parts[1:]) 

128 return encoded.replace("_", "-") 

129 

130 

131def get_claude_config_dir() -> Path: 

132 """Get the Claude config directory, respecting CLAUDE_CONFIG_DIR env var. 

133 

134 Claude SDK uses ~/.claude by default, but this can be overridden 

135 via CLAUDE_CONFIG_DIR environment variable for testing. 

136 

137 Returns: 

138 Path to the Claude config directory. 

139 """ 

140 return Path(os.environ.get("CLAUDE_CONFIG_DIR", str(Path.home() / ".claude"))) 

141 

142 

143def get_claude_log_path(repo_path: Path, session_id: str) -> Path: 

144 """Get path to Claude SDK's session log file. 

145 

146 Claude SDK writes session logs to: 

147 {claude_config_dir}/projects/{encoded-repo-path}/{session_id}.jsonl 

148 

149 The base directory can be overridden via CLAUDE_CONFIG_DIR env var. 

150 

151 Args: 

152 repo_path: Repository path the session was run in. 

153 session_id: Claude SDK session ID (UUID from ResultMessage). 

154 

155 Returns: 

156 Path to the JSONL log file. 

157 """ 

158 encoded = encode_repo_path(repo_path) 

159 return get_claude_config_dir() / "projects" / encoded / f"{session_id}.jsonl" 

160 

161 

162class EnvConfig: 

163 """Environment configuration implementing EnvConfigPort. 

164 

165 Provides access to environment paths for dependency injection into domain modules. 

166 This class wraps the module-level functions to satisfy the EnvConfigPort protocol. 

167 """ 

168 

169 @property 

170 def scripts_dir(self) -> Path: 

171 """Path to the scripts directory (e.g., test-mutex.sh).""" 

172 return SCRIPTS_DIR 

173 

174 @property 

175 def cache_dir(self) -> Path: 

176 """Path to the mala cache directory.""" 

177 return get_cache_dir() 

178 

179 @property 

180 def lock_dir(self) -> Path: 

181 """Path to the lock directory for multi-agent coordination.""" 

182 return get_lock_dir() 

183 

184 def find_cerberus_bin_path(self) -> Path | None: 

185 """Find the cerberus plugin bin directory. 

186 

187 Returns: 

188 Path to cerberus bin directory, or None if not found. 

189 """ 

190 from src.infra.io.config import _find_cerberus_bin_path 

191 

192 return _find_cerberus_bin_path(get_claude_config_dir())