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

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 DesignTokensConfig: 

270 """Design system token configuration.""" 

271 

272 enabled: bool = True 

273 path: str = ".ll/design-tokens" 

274 primitives_file: str = "primitives.json" 

275 semantic_file: str = "semantic.json" 

276 themes_dir: str = "themes" 

277 active_theme: str = "light" 

278 

279 @classmethod 

280 def from_dict(cls, data: dict[str, Any]) -> DesignTokensConfig: 

281 """Create DesignTokensConfig from dictionary.""" 

282 return cls( 

283 enabled=data.get("enabled", True), 

284 path=data.get("path", ".ll/design-tokens"), 

285 primitives_file=data.get("primitives_file", "primitives.json"), 

286 semantic_file=data.get("semantic_file", "semantic.json"), 

287 themes_dir=data.get("themes_dir", "themes"), 

288 active_theme=data.get("active_theme", "light"), 

289 ) 

290 

291 

292@dataclass 

293class SprintsConfig: 

294 """Sprint management configuration.""" 

295 

296 sprints_dir: str = ".sprints" 

297 default_timeout: int = 3600 

298 default_max_workers: int = 2 

299 

300 @classmethod 

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

302 """Create SprintsConfig from dictionary.""" 

303 return cls( 

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

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

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

307 ) 

308 

309 

310@dataclass 

311class LearningTestsConfig: 

312 """Learning test registry configuration.""" 

313 

314 stale_after_days: int = 30 

315 

316 @classmethod 

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

318 """Create LearningTestsConfig from dictionary.""" 

319 return cls( 

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

321 ) 

322 

323 

324@dataclass 

325class LoopsGlyphsConfig: 

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

327 

328 prompt: str = "\u2726" # ✦ 

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

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

331 mcp_tool: str = "\u26a1" # ⚡ 

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

333 route: str = "\u2443" # ⑃ 

334 parallel: str = "\u2225" # ∥ 

335 

336 @classmethod 

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

338 """Create LoopsGlyphsConfig from dictionary.""" 

339 return cls( 

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

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

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

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

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

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

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

347 ) 

348 

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

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

351 return { 

352 "prompt": self.prompt, 

353 "slash_command": self.slash_command, 

354 "shell": self.shell, 

355 "mcp_tool": self.mcp_tool, 

356 "sub_loop": self.sub_loop, 

357 "route": self.route, 

358 "parallel": self.parallel, 

359 } 

360 

361 

362@dataclass 

363class LoopsConfig: 

364 """FSM loop configuration.""" 

365 

366 loops_dir: str = ".loops" 

367 queue_wait_timeout_seconds: int = 86400 

368 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

369 

370 @classmethod 

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

372 """Create LoopsConfig from dictionary.""" 

373 return cls( 

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

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

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

377 ) 

378 

379 

380@dataclass 

381class GitHubSyncConfig: 

382 """GitHub-specific sync configuration.""" 

383 

384 repo: str | None = None 

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

386 default_factory=lambda: { 

387 "BUG": "bug", 

388 "FEAT": "enhancement", 

389 "ENH": "enhancement", 

390 "EPIC": "epic", 

391 } 

392 ) 

393 priority_labels: bool = True 

394 sync_completed: bool = False 

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

396 pull_template: str = "minimal" 

397 pull_limit: int = 500 

398 

399 @classmethod 

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

401 """Create GitHubSyncConfig from dictionary.""" 

402 return cls( 

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

404 label_mapping=data.get( 

405 "label_mapping", 

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

407 ), 

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

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

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

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

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

413 ) 

414 

415 

416@dataclass 

417class SyncConfig: 

418 """Issue sync configuration.""" 

419 

420 enabled: bool = False 

421 provider: str = "github" 

422 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

423 

424 @classmethod 

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

426 """Create SyncConfig from dictionary.""" 

427 return cls( 

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

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

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

431 ) 

432 

433 

434@dataclass 

435class SocketEventsConfig: 

436 """UnixSocketTransport configuration.""" 

437 

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

439 max_clients: int = 32 

440 

441 @classmethod 

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

443 """Create SocketEventsConfig from dictionary.""" 

444 return cls( 

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

446 max_clients=data.get("max_clients", 32), 

447 ) 

448 

449 

450@dataclass 

451class OTelEventsConfig: 

452 """OTelTransport configuration.""" 

453 

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

455 service_name: str = "little-loops" 

456 

457 @classmethod 

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

459 """Create OTelEventsConfig from dictionary.""" 

460 return cls( 

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

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

463 ) 

464 

465 

466@dataclass 

467class WebhookEventsConfig: 

468 """WebhookTransport configuration.""" 

469 

470 url: str | None = None 

471 batch_ms: int = 1000 

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

473 

474 @classmethod 

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

476 """Create WebhookEventsConfig from dictionary.""" 

477 return cls( 

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

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

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

481 ) 

482 

483 

484@dataclass 

485class SqliteEventsConfig: 

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

487 

488 path: str = ".ll/history.db" 

489 

490 @classmethod 

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

492 """Create SqliteEventsConfig from dictionary.""" 

493 return cls( 

494 path=data.get("path", ".ll/history.db"), 

495 ) 

496 

497 

498@dataclass 

499class EventsConfig: 

500 """Event transport configuration. 

501 

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

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

504 unknown names are skipped with a warning. 

505 """ 

506 

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

508 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

509 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

510 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

511 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

512 

513 @classmethod 

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

515 """Create EventsConfig from dictionary.""" 

516 return cls( 

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

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

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

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

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

522 )