Coverage for little_loops / config / automation.py: 100%

1"""Automation and execution configuration dataclasses. 

2 

3Covers automation scripts, parallel execution, confidence gates, 

4command behavior, and dependency analysis configuration. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import Any 

11 

12 

13@dataclass 

14class AutomationConfig: 

15 """Automation script configuration.""" 

16 

17 timeout_seconds: int = 3600 

18 idle_timeout_seconds: int = 0 # Kill if no output for N seconds (0 to disable) 

19 state_file: str = ".auto-manage-state.json" 

20 worktree_base: str = ".worktrees" 

21 max_workers: int = 2 

22 stream_output: bool = True 

23 max_continuations: int = 3 # Max session restarts on context handoff 

24 

25 @classmethod 

26 def from_dict(cls, data: dict[str, Any]) -> AutomationConfig: 

27 """Create AutomationConfig from dictionary.""" 

28 return cls( 

29 timeout_seconds=data.get("timeout_seconds", 3600), 

30 idle_timeout_seconds=data.get("idle_timeout_seconds", 0), 

31 state_file=data.get("state_file", ".auto-manage-state.json"), 

32 worktree_base=data.get("worktree_base", ".worktrees"), 

33 max_workers=data.get("max_workers", 2), 

34 stream_output=data.get("stream_output", True), 

35 max_continuations=data.get("max_continuations", 3), 

36 ) 

37 

38 

39@dataclass 

40class ParallelAutomationConfig: 

41 """Parallel automation configuration using composition. 

42 

43 Uses AutomationConfig for shared settings (max_workers, worktree_base, 

44 state_file, timeout_seconds, stream_output) plus parallel-specific fields. 

45 """ 

46 

47 base: AutomationConfig 

48 p0_sequential: bool = True 

49 max_merge_retries: int = 2 

50 command_prefix: str = "/ll:" 

51 ready_command: str = "ready-issue {{issue_id}}" 

52 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}" 

53 decide_command: str = "decide-issue {{issue_id}}" 

54 worktree_copy_files: list[str] = field( 

55 default_factory=lambda: [".claude/settings.local.json", ".env"] 

56 ) 

57 require_code_changes: bool = True 

58 use_feature_branches: bool = False 

59 push_feature_branches: bool = False 

60 open_pr_for_feature_branches: bool = False 

61 base_branch: str = "main" 

62 remote_name: str = "origin" 

63 

64 @classmethod 

65 def from_dict(cls, data: dict[str, Any]) -> ParallelAutomationConfig: 

66 """Create ParallelAutomationConfig from dictionary. 

67 

68 Shared fields use parallel-specific defaults: 

69 - state_file: ".parallel-manage-state.json" 

70 - stream_output: False 

71 """ 

72 base = AutomationConfig( 

73 timeout_seconds=data.get("timeout_per_issue", data.get("timeout_seconds", 3600)), 

74 state_file=data.get("state_file", ".parallel-manage-state.json"), 

75 worktree_base=data.get("worktree_base", ".worktrees"), 

76 max_workers=data.get("max_workers", 2), 

77 stream_output=data.get("stream_subprocess_output", data.get("stream_output", False)), 

78 ) 

79 return cls( 

80 base=base, 

81 p0_sequential=data.get("p0_sequential", True), 

82 max_merge_retries=data.get("max_merge_retries", 2), 

83 command_prefix=data.get("command_prefix", "/ll:"), 

84 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"), 

85 manage_command=data.get( 

86 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}" 

87 ), 

88 decide_command=data.get("decide_command", "decide-issue {{issue_id}}"), 

89 worktree_copy_files=data.get( 

90 "worktree_copy_files", [".claude/settings.local.json", ".env"] 

91 ), 

92 require_code_changes=data.get("require_code_changes", True), 

93 use_feature_branches=data.get("use_feature_branches", False), 

94 push_feature_branches=data.get("push_feature_branches", False), 

95 open_pr_for_feature_branches=data.get("open_pr_for_feature_branches", False), 

96 base_branch=data.get("base_branch", "main"), 

97 remote_name=data.get("remote_name", "origin"), 

98 ) 

99 

100 

101@dataclass 

102class ConfidenceGateConfig: 

103 """Confidence score gate configuration for manage-issue.""" 

104 

105 enabled: bool = False 

106 readiness_threshold: int = 85 

107 outcome_threshold: int = 70 

108 

109 @classmethod 

110 def from_dict(cls, data: dict[str, Any]) -> ConfidenceGateConfig: 

111 """Create ConfidenceGateConfig from dictionary.""" 

112 legacy = data.get("threshold", 85) 

113 return cls( 

114 enabled=data.get("enabled", False), 

115 readiness_threshold=data.get("readiness_threshold", legacy), 

116 outcome_threshold=data.get("outcome_threshold", 70), 

117 ) 

118 

119 

120@dataclass 

121class RateLimitsConfig: 

122 """Global rate-limit resilience configuration. 

123 

124 Defaults for per-state rate-limit handling when a state does not 

125 override `rate_limit_max_wait_seconds` or `rate_limit_long_wait_ladder`, 

126 plus shared circuit-breaker coordination settings used across parallel 

127 worktrees. 

128 

129 Attributes: 

130 max_wait_seconds: Total wall-clock budget for rate-limit handling 

131 before routing to `on_rate_limit_exhausted`. Default 21600 (6h). 

132 long_wait_ladder: Backoff ladder (seconds) for the long-wait tier 

133 used once the short-tier retry budget is spent. 

134 circuit_breaker_enabled: Whether the shared circuit breaker is active. 

135 circuit_breaker_path: Path (relative to project root) for the shared 

136 circuit-breaker state file. 

137 """ 

138 

139 max_wait_seconds: int = 21600 

140 long_wait_ladder: list[int] = field(default_factory=lambda: [300, 900, 1800, 3600]) 

141 circuit_breaker_enabled: bool = True 

142 circuit_breaker_path: str = ".loops/tmp/rate-limit-circuit.json" 

143 

144 @classmethod 

145 def from_dict(cls, data: dict[str, Any]) -> RateLimitsConfig: 

146 """Create RateLimitsConfig from dictionary.""" 

147 return cls( 

148 max_wait_seconds=data.get("max_wait_seconds", 21600), 

149 long_wait_ladder=data.get("long_wait_ladder", [300, 900, 1800, 3600]), 

150 circuit_breaker_enabled=data.get("circuit_breaker_enabled", True), 

151 circuit_breaker_path=data.get( 

152 "circuit_breaker_path", ".loops/tmp/rate-limit-circuit.json" 

153 ), 

154 ) 

155 

156 

157@dataclass 

158class RecursiveRefineConfig: 

159 """Configuration for the recursive-refine loop.""" 

160 

161 max_depth: int = 3 

162 

163 @classmethod 

164 def from_dict(cls, data: dict[str, Any]) -> RecursiveRefineConfig: 

165 """Create RecursiveRefineConfig from dictionary.""" 

166 return cls(max_depth=data.get("max_depth", 3)) 

167 

168 

169@dataclass 

170class CommandsConfig: 

171 """Command customization configuration.""" 

172 

173 pre_implement: str | None = None 

174 post_implement: str | None = None 

175 custom_verification: list[str] = field(default_factory=list) 

176 confidence_gate: ConfidenceGateConfig = field(default_factory=ConfidenceGateConfig) 

177 tdd_mode: bool = False 

178 rate_limits: RateLimitsConfig = field(default_factory=RateLimitsConfig) 

179 recursive_refine: RecursiveRefineConfig = field(default_factory=RecursiveRefineConfig) 

180 

181 @classmethod 

182 def from_dict(cls, data: dict[str, Any]) -> CommandsConfig: 

183 """Create CommandsConfig from dictionary.""" 

184 return cls( 

185 pre_implement=data.get("pre_implement"), 

186 post_implement=data.get("post_implement"), 

187 custom_verification=data.get("custom_verification", []), 

188 confidence_gate=ConfidenceGateConfig.from_dict(data.get("confidence_gate", {})), 

189 tdd_mode=data.get("tdd_mode", False), 

190 rate_limits=RateLimitsConfig.from_dict(data.get("rate_limits", {})), 

191 recursive_refine=RecursiveRefineConfig.from_dict(data.get("recursive_refine", {})), 

192 ) 

193 

194 

195@dataclass 

196class ScoringWeightsConfig: 

197 """Scoring weights for semantic conflict analysis. 

198 

199 Weights for the three signals used in compute_conflict_score(). 

200 Should sum to 1.0 for normalized scoring. 

201 

202 Attributes: 

203 semantic: Weight for semantic target overlap (component/function names) 

204 section: Weight for section mention overlap (UI regions) 

205 type: Weight for modification type match 

206 """ 

207 

208 semantic: float = 0.5 

209 section: float = 0.3 

210 type: float = 0.2 

211 

212 @classmethod 

213 def from_dict(cls, data: dict[str, Any]) -> ScoringWeightsConfig: 

214 """Create ScoringWeightsConfig from dictionary.""" 

215 return cls( 

216 semantic=data.get("semantic", 0.5), 

217 section=data.get("section", 0.3), 

218 type=data.get("type", 0.2), 

219 ) 

220 

221 

222@dataclass 

223class DependencyMappingConfig: 

224 """Dependency mapping threshold configuration. 

225 

226 Controls overlap detection sensitivity and conflict scoring thresholds. 

227 Default values match the previously hardcoded constants for backwards 

228 compatibility. 

229 

230 Attributes: 

231 overlap_min_files: Minimum overlapping files to trigger overlap 

232 overlap_min_ratio: Minimum ratio of overlapping files to smaller set 

233 min_directory_depth: Minimum path segments for directory overlap 

234 conflict_threshold: Below = parallel-safe, above = dependency proposed 

235 high_conflict_threshold: Above = HIGH conflict label 

236 confidence_modifier: Applied when dependency direction is ambiguous 

237 scoring_weights: Weights for semantic/section/type signals 

238 exclude_common_files: Infrastructure files excluded from overlap detection 

239 """ 

240 

241 overlap_min_files: int = 2 

242 overlap_min_ratio: float = 0.25 

243 min_directory_depth: int = 2 

244 conflict_threshold: float = 0.4 

245 high_conflict_threshold: float = 0.7 

246 confidence_modifier: float = 0.5 

247 scoring_weights: ScoringWeightsConfig = field(default_factory=ScoringWeightsConfig) 

248 exclude_common_files: list[str] = field( 

249 default_factory=lambda: [ 

250 "__init__.py", 

251 "pyproject.toml", 

252 "setup.py", 

253 "setup.cfg", 

254 "CHANGELOG.md", 

255 "README.md", 

256 "conftest.py", 

257 ] 

258 ) 

259 

260 @classmethod 

261 def from_dict(cls, data: dict[str, Any]) -> DependencyMappingConfig: 

262 """Create DependencyMappingConfig from dictionary.""" 

263 return cls( 

264 overlap_min_files=data.get("overlap_min_files", 2), 

265 overlap_min_ratio=data.get("overlap_min_ratio", 0.25), 

266 min_directory_depth=data.get("min_directory_depth", 2), 

267 conflict_threshold=data.get("conflict_threshold", 0.4), 

268 high_conflict_threshold=data.get("high_conflict_threshold", 0.7), 

269 confidence_modifier=data.get("confidence_modifier", 0.5), 

270 scoring_weights=ScoringWeightsConfig.from_dict(data.get("scoring_weights", {})), 

271 exclude_common_files=data.get( 

272 "exclude_common_files", 

273 [ 

274 "__init__.py", 

275 "pyproject.toml", 

276 "setup.py", 

277 "setup.cfg", 

278 "CHANGELOG.md", 

279 "README.md", 

280 "conftest.py", 

281 ], 

282 ), 

283 )