Coverage for little_loops / config / features.py: 75%

1"""Feature-related configuration dataclasses. 

2 

3Covers issue tracking, scanning, sprint management, loop management, 

4and sync configuration. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import Any 

11 

12# Required categories that must always exist (cannot be removed by user config) 

13REQUIRED_CATEGORIES: dict[str, dict[str, str]] = { 

14 "bugs": {"prefix": "BUG", "dir": "bugs", "action": "fix"}, 

15 "features": {"prefix": "FEAT", "dir": "features", "action": "implement"}, 

16 "enhancements": {"prefix": "ENH", "dir": "enhancements", "action": "improve"}, 

17 "epics": {"prefix": "EPIC", "dir": "epics", "action": "coordinate"}, 

18} 

19 

20# Default categories (same as required by default, could include optional defaults) 

21DEFAULT_CATEGORIES: dict[str, dict[str, str]] = { 

22 **REQUIRED_CATEGORIES, 

23} 

24 

25 

26@dataclass 

27class CategoryConfig: 

28 """Configuration for an issue category.""" 

29 

30 prefix: str 

31 dir: str 

32 action: str = "fix" 

33 

34 @classmethod 

35 def from_dict(cls, key: str, data: dict[str, Any]) -> CategoryConfig: 

36 """Create CategoryConfig from dictionary.""" 

37 return cls( 

38 prefix=data.get("prefix", key.upper()[:3]), 

39 dir=data.get("dir", key), 

40 action=data.get("action", "fix"), 

41 ) 

42 

43 

44@dataclass 

45class DuplicateDetectionConfig: 

46 """Thresholds for duplicate issue detection.""" 

47 

48 exact_threshold: float = 0.8 

49 similar_threshold: float = 0.5 

50 

51 @classmethod 

52 def from_dict(cls, data: dict[str, Any]) -> DuplicateDetectionConfig: 

53 """Create DuplicateDetectionConfig from dictionary.""" 

54 return cls( 

55 exact_threshold=data.get("exact_threshold", 0.8), 

56 similar_threshold=data.get("similar_threshold", 0.5), 

57 ) 

58 

59 

60VALID_NEXT_ISSUE_STRATEGIES: frozenset[str] = frozenset({"confidence_first", "priority_first"}) 

61VALID_NEXT_ISSUE_SORT_KEYS: frozenset[str] = frozenset( 

62 { 

63 "priority", 

64 "outcome_confidence", 

65 "confidence_score", 

66 "effort", 

67 "impact", 

68 "score_complexity", 

69 "score_test_coverage", 

70 "score_ambiguity", 

71 "score_change_surface", 

72 } 

73) 

74VALID_NEXT_ISSUE_SORT_DIRECTIONS: frozenset[str] = frozenset({"asc", "desc"}) 

75 

76 

77@dataclass 

78class NextIssueSortKey: 

79 """A single (key, direction) pair for custom next-issue sort orderings.""" 

80 

81 key: str 

82 direction: str = "asc" 

83 

84 @classmethod 

85 def from_dict(cls, data: dict[str, Any]) -> NextIssueSortKey: 

86 """Create NextIssueSortKey from dictionary, validating key/direction.""" 

87 key = data.get("key") 

88 if key not in VALID_NEXT_ISSUE_SORT_KEYS: 

89 raise ValueError(f"Unknown sort key: {key!r}") 

90 direction = data.get("direction", "asc") 

91 if direction not in VALID_NEXT_ISSUE_SORT_DIRECTIONS: 

92 raise ValueError(f"Unknown sort direction: {direction!r}") 

93 return cls(key=key, direction=direction) 

94 

95 

96@dataclass 

97class NextIssueConfig: 

98 """Selection behavior for ll-issues next-issue / next-issues commands. 

99 

100 Strategy presets: 

101 - "confidence_first" (default): sort by (-outcome_confidence, -confidence_score, priority_int) 

102 - "priority_first": sort by (priority_int, -outcome_confidence, -confidence_score) 

103 

104 If sort_keys is provided, it overrides strategy. 

105 """ 

106 

107 strategy: str = "confidence_first" 

108 sort_keys: list[NextIssueSortKey] | None = None 

109 

110 @classmethod 

111 def from_dict(cls, data: dict[str, Any]) -> NextIssueConfig: 

112 """Create NextIssueConfig from dictionary, validating strategy and sort_keys.""" 

113 strategy = data.get("strategy", "confidence_first") 

114 if strategy not in VALID_NEXT_ISSUE_STRATEGIES: 

115 raise ValueError(f"Unknown strategy: {strategy!r}") 

116 sort_keys_data = data.get("sort_keys") 

117 sort_keys: list[NextIssueSortKey] | None 

118 if sort_keys_data is None: 

119 sort_keys = None 

120 else: 

121 sort_keys = [NextIssueSortKey.from_dict(entry) for entry in sort_keys_data] 

122 return cls(strategy=strategy, sort_keys=sort_keys) 

123 

124 

125@dataclass 

126class IssuesConfig: 

127 """Issue management configuration.""" 

128 

129 base_dir: str = ".issues" 

130 categories: dict[str, CategoryConfig] = field(default_factory=dict) 

131 completed_dir: str = "completed" # DEPRECATED: use IssueInfo.status instead 

132 deferred_dir: str = "deferred" # DEPRECATED: use IssueInfo.status instead 

133 priorities: list[str] = field(default_factory=lambda: ["P0", "P1", "P2", "P3", "P4", "P5"]) 

134 templates_dir: str | None = None 

135 capture_template: str = "full" 

136 duplicate_detection: DuplicateDetectionConfig = field(default_factory=DuplicateDetectionConfig) 

137 next_issue: NextIssueConfig = field(default_factory=NextIssueConfig) 

138 

139 @classmethod 

140 def from_dict(cls, data: dict[str, Any]) -> IssuesConfig: 

141 """Create IssuesConfig from dictionary. 

142 

143 Required categories (bugs, features, enhancements) are automatically 

144 included if not specified in user config. 

145 """ 

146 # Start with user categories or empty dict 

147 categories_data = dict(data.get("categories", {})) 

148 

149 # Ensure required categories exist (merge with defaults) 

150 for key, defaults in REQUIRED_CATEGORIES.items(): 

151 if key not in categories_data: 

152 categories_data[key] = defaults 

153 

154 categories = { 

155 key: CategoryConfig.from_dict(key, value) for key, value in categories_data.items() 

156 } 

157 return cls( 

158 base_dir=data.get("base_dir", ".issues"), 

159 categories=categories, 

160 completed_dir=data.get( 

161 "completed_dir", "completed" 

162 ), # deprecated: kept for backward compat 

163 deferred_dir=data.get( 

164 "deferred_dir", "deferred" 

165 ), # deprecated: kept for backward compat 

166 priorities=data.get("priorities", ["P0", "P1", "P2", "P3", "P4", "P5"]), 

167 templates_dir=data.get("templates_dir"), 

168 capture_template=data.get("capture_template", "full"), 

169 duplicate_detection=DuplicateDetectionConfig.from_dict( 

170 data.get("duplicate_detection", {}) 

171 ), 

172 next_issue=NextIssueConfig.from_dict(data.get("next_issue", {})), 

173 ) 

174 

175 def get_category_by_prefix(self, prefix: str) -> CategoryConfig | None: 

176 """Get category config by prefix (e.g., 'BUG', 'FEAT'). 

177 

178 Args: 

179 prefix: Issue type prefix to look up 

180 

181 Returns: 

182 CategoryConfig if found, None otherwise 

183 """ 

184 for category in self.categories.values(): 

185 if category.prefix == prefix: 

186 return category 

187 return None 

188 

189 def get_category_by_dir(self, dir_name: str) -> CategoryConfig | None: 

190 """Get category config by directory name. 

191 

192 Args: 

193 dir_name: Directory name to look up 

194 

195 Returns: 

196 CategoryConfig if found, None otherwise 

197 """ 

198 for category in self.categories.values(): 

199 if category.dir == dir_name: 

200 return category 

201 return None 

202 

203 def get_all_prefixes(self) -> list[str]: 

204 """Get all configured issue type prefixes. 

205 

206 Returns: 

207 List of prefixes (e.g., ['BUG', 'FEAT', 'ENH']) 

208 """ 

209 return [cat.prefix for cat in self.categories.values()] 

210 

211 def get_all_dirs(self) -> list[str]: 

212 """Get all configured issue directory names. 

213 

214 Returns: 

215 List of directory names (e.g., ['bugs', 'features', 'enhancements']) 

216 """ 

217 return [cat.dir for cat in self.categories.values()] 

218 

219 

220@dataclass 

221class ScanConfig: 

222 """Codebase scanning configuration.""" 

223 

224 focus_dirs: list[str] = field(default_factory=lambda: ["src/", "tests/"]) 

225 exclude_patterns: list[str] = field( 

226 default_factory=lambda: ["**/node_modules/**", "**/__pycache__/**", "**/.git/**"] 

227 ) 

228 custom_agents: list[str] = field(default_factory=list) 

229 

230 @classmethod 

231 def from_dict(cls, data: dict[str, Any]) -> ScanConfig: 

232 """Create ScanConfig from dictionary.""" 

233 return cls( 

234 focus_dirs=data.get("focus_dirs", ["src/", "tests/"]), 

235 exclude_patterns=data.get( 

236 "exclude_patterns", 

237 ["**/node_modules/**", "**/__pycache__/**", "**/.git/**"], 

238 ), 

239 custom_agents=data.get("custom_agents", []), 

240 ) 

241 

242 

243@dataclass 

244class SprintsConfig: 

245 """Sprint management configuration.""" 

246 

247 sprints_dir: str = ".sprints" 

248 default_timeout: int = 3600 

249 default_max_workers: int = 2 

250 

251 @classmethod 

252 def from_dict(cls, data: dict[str, Any]) -> SprintsConfig: 

253 """Create SprintsConfig from dictionary.""" 

254 return cls( 

255 sprints_dir=data.get("sprints_dir", ".sprints"), 

256 default_timeout=data.get("default_timeout", 3600), 

257 default_max_workers=data.get("default_max_workers", 2), 

258 ) 

259 

260 

261@dataclass 

262class LearningTestsConfig: 

263 """Learning test registry configuration.""" 

264 

265 stale_after_days: int = 30 

266 

267 @classmethod 

268 def from_dict(cls, data: dict[str, Any]) -> LearningTestsConfig: 

269 """Create LearningTestsConfig from dictionary.""" 

270 return cls( 

271 stale_after_days=data.get("stale_after_days", 30), 

272 ) 

273 

274 

275@dataclass 

276class LoopsGlyphsConfig: 

277 """Unicode badge/glyph overrides for FSM box diagram state badges.""" 

278 

279 prompt: str = "\u2726" # ✦ 

280 slash_command: str = "/\u2501\u25ba" # /━► 

281 shell: str = "\u276f_" # ❯_ 

282 mcp_tool: str = "\u26a1" # ⚡ 

283 sub_loop: str = "\u21b3\u27f3" # ↳⟳ 

284 route: str = "\u2443" # ⑃ 

285 parallel: str = "\u2225" # ∥ 

286 

287 @classmethod 

288 def from_dict(cls, data: dict[str, Any]) -> LoopsGlyphsConfig: 

289 """Create LoopsGlyphsConfig from dictionary.""" 

290 return cls( 

291 prompt=data.get("prompt", "\u2726"), 

292 slash_command=data.get("slash_command", "/\u2501\u25ba"), 

293 shell=data.get("shell", "\u276f_"), 

294 mcp_tool=data.get("mcp_tool", "\u26a1"), 

295 sub_loop=data.get("sub_loop", "\u21b3\u27f3"), 

296 route=data.get("route", "\u2443"), 

297 parallel=data.get("parallel", "\u2225"), 

298 ) 

299 

300 def to_dict(self) -> dict[str, str]: 

301 """Convert to a glyph-key→string dict for use by _get_state_badge.""" 

302 return { 

303 "prompt": self.prompt, 

304 "slash_command": self.slash_command, 

305 "shell": self.shell, 

306 "mcp_tool": self.mcp_tool, 

307 "sub_loop": self.sub_loop, 

308 "route": self.route, 

309 "parallel": self.parallel, 

310 } 

311 

312 

313@dataclass 

314class LoopsConfig: 

315 """FSM loop configuration.""" 

316 

317 loops_dir: str = ".loops" 

318 queue_wait_timeout_seconds: int = 86400 

319 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

320 

321 @classmethod 

322 def from_dict(cls, data: dict[str, Any]) -> LoopsConfig: 

323 """Create LoopsConfig from dictionary.""" 

324 return cls( 

325 loops_dir=data.get("loops_dir", ".loops"), 

326 queue_wait_timeout_seconds=data.get("queue_wait_timeout_seconds", 86400), 

327 glyphs=LoopsGlyphsConfig.from_dict(data.get("glyphs", {})), 

328 ) 

329 

330 

331@dataclass 

332class GitHubSyncConfig: 

333 """GitHub-specific sync configuration.""" 

334 

335 repo: str | None = None 

336 label_mapping: dict[str, str] = field( 

337 default_factory=lambda: { 

338 "BUG": "bug", 

339 "FEAT": "enhancement", 

340 "ENH": "enhancement", 

341 "EPIC": "epic", 

342 } 

343 ) 

344 priority_labels: bool = True 

345 sync_completed: bool = False 

346 state_file: str = ".ll/ll-sync-state.json" 

347 pull_template: str = "minimal" 

348 pull_limit: int = 500 

349 

350 @classmethod 

351 def from_dict(cls, data: dict[str, Any]) -> GitHubSyncConfig: 

352 """Create GitHubSyncConfig from dictionary.""" 

353 return cls( 

354 repo=data.get("repo"), 

355 label_mapping=data.get( 

356 "label_mapping", 

357 {"BUG": "bug", "FEAT": "enhancement", "ENH": "enhancement", "EPIC": "epic"}, 

358 ), 

359 priority_labels=data.get("priority_labels", True), 

360 sync_completed=data.get("sync_completed", False), 

361 state_file=data.get("state_file", ".ll/ll-sync-state.json"), 

362 pull_template=data.get("pull_template", "minimal"), 

363 pull_limit=data.get("pull_limit", 500), 

364 ) 

365 

366 

367@dataclass 

368class SyncConfig: 

369 """Issue sync configuration.""" 

370 

371 enabled: bool = False 

372 provider: str = "github" 

373 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

374 

375 @classmethod 

376 def from_dict(cls, data: dict[str, Any]) -> SyncConfig: 

377 """Create SyncConfig from dictionary.""" 

378 return cls( 

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

380 provider=data.get("provider", "github"), 

381 github=GitHubSyncConfig.from_dict(data.get("github", {})), 

382 ) 

383 

384 

385@dataclass 

386class SocketEventsConfig: 

387 """UnixSocketTransport configuration.""" 

388 

389 path: str = ".ll/events.sock" 

390 max_clients: int = 8 

391 

392 @classmethod 

393 def from_dict(cls, data: dict[str, Any]) -> SocketEventsConfig: 

394 """Create SocketEventsConfig from dictionary.""" 

395 return cls( 

396 path=data.get("path", ".ll/events.sock"), 

397 max_clients=data.get("max_clients", 8), 

398 ) 

399 

400 

401@dataclass 

402class OTelEventsConfig: 

403 """OTelTransport configuration.""" 

404 

405 endpoint: str = "http://localhost:4317" 

406 service_name: str = "little-loops" 

407 

408 @classmethod 

409 def from_dict(cls, data: dict[str, Any]) -> OTelEventsConfig: 

410 """Create OTelEventsConfig from dictionary.""" 

411 return cls( 

412 endpoint=data.get("endpoint", "http://localhost:4317"), 

413 service_name=data.get("service_name", "little-loops"), 

414 ) 

415 

416 

417@dataclass 

418class WebhookEventsConfig: 

419 """WebhookTransport configuration.""" 

420 

421 url: str | None = None 

422 batch_ms: int = 1000 

423 headers: dict[str, str] = field(default_factory=dict) 

424 

425 @classmethod 

426 def from_dict(cls, data: dict[str, Any]) -> WebhookEventsConfig: 

427 """Create WebhookEventsConfig from dictionary.""" 

428 return cls( 

429 url=data.get("url", None), 

430 batch_ms=data.get("batch_ms", 1000), 

431 headers=data.get("headers", {}), 

432 ) 

433 

434 

435@dataclass 

436class EventsConfig: 

437 """Event transport configuration. 

438 

439 Lists the transports to wire onto the EventBus at runtime. Names are 

440 resolved against the registry in `little_loops.transport.wire_transports`; 

441 unknown names are skipped with a warning. 

442 """ 

443 

444 transports: list[str] = field(default_factory=list) 

445 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

446 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

447 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

448 

449 @classmethod 

450 def from_dict(cls, data: dict[str, Any]) -> EventsConfig: 

451 """Create EventsConfig from dictionary.""" 

452 return cls( 

453 transports=data.get("transports", []), 

454 socket=SocketEventsConfig.from_dict(data.get("socket", {})), 

455 otel=OTelEventsConfig.from_dict(data.get("otel", {})), 

456 webhook=WebhookEventsConfig.from_dict(data.get("webhook", {})), 

457 )