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

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 

13def feature_enabled(config_data: dict[str, Any], dot_path: str) -> bool: 

14 """Return whether the boolean flag at *dot_path* is enabled in *config_data*. 

15 

16 Python port of ``hooks/scripts/lib/common.sh:ll_feature_enabled``. Operates 

17 on an already-parsed config dict (the bash version uses ``jq`` on the file). 

18 Mirrors jq's ``// false`` default: missing keys, non-dict intermediates, or 

19 non-truthy terminal values all yield ``False``. 

20 

21 Examples: 

22 >>> feature_enabled({"context_monitor": {"enabled": True}}, "context_monitor.enabled") 

23 True 

24 >>> feature_enabled({"context_monitor": {"enabled": False}}, "context_monitor.enabled") 

25 False 

26 >>> feature_enabled({}, "sync.enabled") 

27 False 

28 """ 

29 value: Any = config_data 

30 for part in dot_path.split("."): 

31 if not isinstance(value, dict) or part not in value: 

32 return False 

33 value = value[part] 

34 return bool(value) 

35 

36 

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

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

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

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

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

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

43} 

44 

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

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

47 **REQUIRED_CATEGORIES, 

48} 

49 

50 

51@dataclass 

52class CategoryConfig: 

53 """Configuration for an issue category.""" 

54 

55 prefix: str 

56 dir: str 

57 action: str = "fix" 

58 

59 @classmethod 

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

61 """Create CategoryConfig from dictionary.""" 

62 return cls( 

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

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

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

66 ) 

67 

68 

69@dataclass 

70class DuplicateDetectionConfig: 

71 """Thresholds for duplicate issue detection.""" 

72 

73 exact_threshold: float = 0.8 

74 similar_threshold: float = 0.5 

75 

76 @classmethod 

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

78 """Create DuplicateDetectionConfig from dictionary.""" 

79 return cls( 

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

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

82 ) 

83 

84 

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

86VALID_NEXT_ISSUE_SORT_KEYS: frozenset[str] = frozenset( 

87 { 

88 "priority", 

89 "outcome_confidence", 

90 "confidence_score", 

91 "effort", 

92 "impact", 

93 "score_complexity", 

94 "score_test_coverage", 

95 "score_ambiguity", 

96 "score_change_surface", 

97 } 

98) 

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

100 

101 

102@dataclass 

103class NextIssueSortKey: 

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

105 

106 key: str 

107 direction: str = "asc" 

108 

109 @classmethod 

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

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

112 key = data.get("key") 

113 if key not in VALID_NEXT_ISSUE_SORT_KEYS: 

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

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

116 if direction not in VALID_NEXT_ISSUE_SORT_DIRECTIONS: 

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

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

119 

120 

121@dataclass 

122class NextIssueConfig: 

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

124 

125 Strategy presets: 

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

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

128 

129 If sort_keys is provided, it overrides strategy. 

130 """ 

131 

132 strategy: str = "confidence_first" 

133 sort_keys: list[NextIssueSortKey] | None = None 

134 

135 @classmethod 

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

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

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

139 if strategy not in VALID_NEXT_ISSUE_STRATEGIES: 

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

141 sort_keys_data = data.get("sort_keys") 

142 sort_keys: list[NextIssueSortKey] | None 

143 if sort_keys_data is None: 

144 sort_keys = None 

145 else: 

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

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

148 

149 

150@dataclass 

151class IssuesConfig: 

152 """Issue management configuration.""" 

153 

154 base_dir: str = ".issues" 

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

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

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

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

159 templates_dir: str | None = None 

160 capture_template: str = "full" 

161 duplicate_detection: DuplicateDetectionConfig = field(default_factory=DuplicateDetectionConfig) 

162 next_issue: NextIssueConfig = field(default_factory=NextIssueConfig) 

163 

164 @classmethod 

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

166 """Create IssuesConfig from dictionary. 

167 

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

169 included if not specified in user config. 

170 """ 

171 # Start with user categories or empty dict 

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

173 

174 # Ensure required categories exist (merge with defaults) 

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

176 if key not in categories_data: 

177 categories_data[key] = defaults 

178 

179 categories = { 

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

181 } 

182 return cls( 

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

184 categories=categories, 

185 completed_dir=data.get( 

186 "completed_dir", "completed" 

187 ), # deprecated: kept for backward compat 

188 deferred_dir=data.get( 

189 "deferred_dir", "deferred" 

190 ), # deprecated: kept for backward compat 

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

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

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

194 duplicate_detection=DuplicateDetectionConfig.from_dict( 

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

196 ), 

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

198 ) 

199 

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

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

202 

203 Args: 

204 prefix: Issue type prefix to look up 

205 

206 Returns: 

207 CategoryConfig if found, None otherwise 

208 """ 

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

210 if category.prefix == prefix: 

211 return category 

212 return None 

213 

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

215 """Get category config by directory name. 

216 

217 Args: 

218 dir_name: Directory name to look up 

219 

220 Returns: 

221 CategoryConfig if found, None otherwise 

222 """ 

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

224 if category.dir == dir_name: 

225 return category 

226 return None 

227 

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

229 """Get all configured issue type prefixes. 

230 

231 Returns: 

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

233 """ 

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

235 

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

237 """Get all configured issue directory names. 

238 

239 Returns: 

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

241 """ 

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

243 

244 

245@dataclass 

246class ScanConfig: 

247 """Codebase scanning configuration.""" 

248 

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

250 exclude_patterns: list[str] = field( 

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

252 ) 

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

254 

255 @classmethod 

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

257 """Create ScanConfig from dictionary.""" 

258 return cls( 

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

260 exclude_patterns=data.get( 

261 "exclude_patterns", 

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

263 ), 

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

265 ) 

266 

267 

268@dataclass 

269class SprintsConfig: 

270 """Sprint management configuration.""" 

271 

272 sprints_dir: str = ".sprints" 

273 default_timeout: int = 3600 

274 default_max_workers: int = 2 

275 

276 @classmethod 

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

278 """Create SprintsConfig from dictionary.""" 

279 return cls( 

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

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

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

283 ) 

284 

285 

286@dataclass 

287class LearningTestsConfig: 

288 """Learning test registry configuration.""" 

289 

290 stale_after_days: int = 30 

291 

292 @classmethod 

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

294 """Create LearningTestsConfig from dictionary.""" 

295 return cls( 

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

297 ) 

298 

299 

300@dataclass 

301class LoopsGlyphsConfig: 

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

303 

304 prompt: str = "\u2726" # ✦ 

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

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

307 mcp_tool: str = "\u26a1" # ⚡ 

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

309 route: str = "\u2443" # ⑃ 

310 parallel: str = "\u2225" # ∥ 

311 

312 @classmethod 

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

314 """Create LoopsGlyphsConfig from dictionary.""" 

315 return cls( 

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

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

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

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

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

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

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

323 ) 

324 

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

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

327 return { 

328 "prompt": self.prompt, 

329 "slash_command": self.slash_command, 

330 "shell": self.shell, 

331 "mcp_tool": self.mcp_tool, 

332 "sub_loop": self.sub_loop, 

333 "route": self.route, 

334 "parallel": self.parallel, 

335 } 

336 

337 

338@dataclass 

339class LoopsConfig: 

340 """FSM loop configuration.""" 

341 

342 loops_dir: str = ".loops" 

343 queue_wait_timeout_seconds: int = 86400 

344 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

345 

346 @classmethod 

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

348 """Create LoopsConfig from dictionary.""" 

349 return cls( 

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

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

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

353 ) 

354 

355 

356@dataclass 

357class GitHubSyncConfig: 

358 """GitHub-specific sync configuration.""" 

359 

360 repo: str | None = None 

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

362 default_factory=lambda: { 

363 "BUG": "bug", 

364 "FEAT": "enhancement", 

365 "ENH": "enhancement", 

366 "EPIC": "epic", 

367 } 

368 ) 

369 priority_labels: bool = True 

370 sync_completed: bool = False 

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

372 pull_template: str = "minimal" 

373 pull_limit: int = 500 

374 

375 @classmethod 

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

377 """Create GitHubSyncConfig from dictionary.""" 

378 return cls( 

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

380 label_mapping=data.get( 

381 "label_mapping", 

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

383 ), 

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

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

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

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

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

389 ) 

390 

391 

392@dataclass 

393class SyncConfig: 

394 """Issue sync configuration.""" 

395 

396 enabled: bool = False 

397 provider: str = "github" 

398 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

399 

400 @classmethod 

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

402 """Create SyncConfig from dictionary.""" 

403 return cls( 

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

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

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

407 ) 

408 

409 

410@dataclass 

411class SocketEventsConfig: 

412 """UnixSocketTransport configuration.""" 

413 

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

415 max_clients: int = 8 

416 

417 @classmethod 

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

419 """Create SocketEventsConfig from dictionary.""" 

420 return cls( 

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

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

423 ) 

424 

425 

426@dataclass 

427class OTelEventsConfig: 

428 """OTelTransport configuration.""" 

429 

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

431 service_name: str = "little-loops" 

432 

433 @classmethod 

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

435 """Create OTelEventsConfig from dictionary.""" 

436 return cls( 

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

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

439 ) 

440 

441 

442@dataclass 

443class WebhookEventsConfig: 

444 """WebhookTransport configuration.""" 

445 

446 url: str | None = None 

447 batch_ms: int = 1000 

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

449 

450 @classmethod 

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

452 """Create WebhookEventsConfig from dictionary.""" 

453 return cls( 

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

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

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

457 ) 

458 

459 

460@dataclass 

461class SqliteEventsConfig: 

462 """SQLiteTransport configuration (unified session store, FEAT-1112).""" 

463 

464 path: str = ".ll/session.db" 

465 

466 @classmethod 

467 def from_dict(cls, data: dict[str, Any]) -> SqliteEventsConfig: 

468 """Create SqliteEventsConfig from dictionary.""" 

469 return cls( 

470 path=data.get("path", ".ll/session.db"), 

471 ) 

472 

473 

474@dataclass 

475class EventsConfig: 

476 """Event transport configuration. 

477 

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

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

480 unknown names are skipped with a warning. 

481 """ 

482 

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

484 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

485 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

486 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

487 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

488 

489 @classmethod 

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

491 """Create EventsConfig from dictionary.""" 

492 return cls( 

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

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

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

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

497 sqlite=SqliteEventsConfig.from_dict(data.get("sqlite", {})), 

498 )