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 

9import fnmatch 

10from dataclasses import dataclass, field 

11from typing import Any 

12 

13 

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

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

16 

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

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

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

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

21 

22 Examples: 

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

24 True 

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

26 False 

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

28 False 

29 """ 

30 value: Any = config_data 

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

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

33 return False 

34 value = value[part] 

35 return bool(value) 

36 

37 

38def feature_enabled_for( 

39 config_data: dict[str, Any], dot_path: str, subject: str, default: bool = True 

40) -> bool: 

41 """Return whether *subject* matches the glob-pattern list at *dot_path* in *config_data*. 

42 

43 Operates on a raw config dict (same as ``feature_enabled``). Resolves the value at 

44 *dot_path*; if absent or the intermediate path doesn't exist, returns *default*. 

45 The resolved value is normalised to a ``list[str]`` (bare string wrapped in a list, 

46 ``None`` treated as ``["*"]`` — match all) then tested with ``fnmatch.fnmatch``. 

47 

48 Examples: 

49 >>> feature_enabled_for({"analytics": {"capture": {"skills": ["*"]}}}, 

50 ... "analytics.capture.skills", "my-skill") 

51 True 

52 >>> feature_enabled_for({"analytics": {"capture": {"skills": ["Read"]}}}, 

53 ... "analytics.capture.skills", "Write") 

54 False 

55 >>> feature_enabled_for({}, "analytics.capture.skills", "any") 

56 True 

57 """ 

58 value: Any = config_data 

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

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

61 return default 

62 value = value[part] 

63 

64 # Normalise to list[str]: None → match-all, str → [str], list used as-is 

65 if value is None: 

66 patterns: list[str] = ["*"] 

67 elif isinstance(value, str): 

68 patterns = [value] 

69 else: 

70 patterns = list(value) 

71 

72 if not patterns: 

73 return default 

74 

75 return any(fnmatch.fnmatch(subject, p) for p in patterns) 

76 

77 

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

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

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

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

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

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

84} 

85 

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

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

88 **REQUIRED_CATEGORIES, 

89} 

90 

91 

92@dataclass 

93class CategoryConfig: 

94 """Configuration for an issue category.""" 

95 

96 prefix: str 

97 dir: str 

98 action: str = "fix" 

99 

100 @classmethod 

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

102 """Create CategoryConfig from dictionary.""" 

103 return cls( 

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

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

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

107 ) 

108 

109 

110@dataclass 

111class DuplicateDetectionConfig: 

112 """Thresholds for duplicate issue detection.""" 

113 

114 exact_threshold: float = 0.8 

115 similar_threshold: float = 0.5 

116 

117 @classmethod 

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

119 """Create DuplicateDetectionConfig from dictionary.""" 

120 return cls( 

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

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

123 ) 

124 

125 

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

127VALID_NEXT_ISSUE_SORT_KEYS: frozenset[str] = frozenset( 

128 { 

129 "priority", 

130 "outcome_confidence", 

131 "confidence_score", 

132 "effort", 

133 "impact", 

134 "score_complexity", 

135 "score_test_coverage", 

136 "score_ambiguity", 

137 "score_change_surface", 

138 } 

139) 

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

141 

142 

143@dataclass 

144class NextIssueSortKey: 

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

146 

147 key: str 

148 direction: str = "asc" 

149 

150 @classmethod 

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

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

153 key = data.get("key") 

154 if key not in VALID_NEXT_ISSUE_SORT_KEYS: 

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

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

157 if direction not in VALID_NEXT_ISSUE_SORT_DIRECTIONS: 

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

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

160 

161 

162@dataclass 

163class NextIssueConfig: 

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

165 

166 Strategy presets: 

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

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

169 

170 If sort_keys is provided, it overrides strategy. 

171 """ 

172 

173 strategy: str = "confidence_first" 

174 sort_keys: list[NextIssueSortKey] | None = None 

175 

176 @classmethod 

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

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

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

180 if strategy not in VALID_NEXT_ISSUE_STRATEGIES: 

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

182 sort_keys_data = data.get("sort_keys") 

183 sort_keys: list[NextIssueSortKey] | None 

184 if sort_keys_data is None: 

185 sort_keys = None 

186 else: 

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

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

189 

190 

191@dataclass 

192class IssuesConfig: 

193 """Issue management configuration.""" 

194 

195 base_dir: str = ".issues" 

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

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

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

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

200 templates_dir: str | None = None 

201 capture_template: str = "full" 

202 duplicate_detection: DuplicateDetectionConfig = field(default_factory=DuplicateDetectionConfig) 

203 next_issue: NextIssueConfig = field(default_factory=NextIssueConfig) 

204 auto_commit: bool = False 

205 auto_commit_prefix: str = "chore(issues)" 

206 

207 @classmethod 

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

209 """Create IssuesConfig from dictionary. 

210 

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

212 included if not specified in user config. 

213 """ 

214 # Start with user categories or empty dict 

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

216 

217 # Ensure required categories exist (merge with defaults) 

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

219 if key not in categories_data: 

220 categories_data[key] = defaults 

221 

222 categories = { 

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

224 } 

225 return cls( 

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

227 categories=categories, 

228 completed_dir=data.get( 

229 "completed_dir", "completed" 

230 ), # deprecated: kept for backward compat 

231 deferred_dir=data.get( 

232 "deferred_dir", "deferred" 

233 ), # deprecated: kept for backward compat 

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

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

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

237 duplicate_detection=DuplicateDetectionConfig.from_dict( 

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

239 ), 

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

241 auto_commit=data.get("auto_commit", False), 

242 auto_commit_prefix=data.get("auto_commit_prefix", "chore(issues)"), 

243 ) 

244 

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

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

247 

248 Args: 

249 prefix: Issue type prefix to look up 

250 

251 Returns: 

252 CategoryConfig if found, None otherwise 

253 """ 

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

255 if category.prefix == prefix: 

256 return category 

257 return None 

258 

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

260 """Get category config by directory name. 

261 

262 Args: 

263 dir_name: Directory name to look up 

264 

265 Returns: 

266 CategoryConfig if found, None otherwise 

267 """ 

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

269 if category.dir == dir_name: 

270 return category 

271 return None 

272 

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

274 """Get all configured issue type prefixes. 

275 

276 Returns: 

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

278 """ 

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

280 

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

282 """Get all configured issue directory names. 

283 

284 Returns: 

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

286 """ 

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

288 

289 

290@dataclass 

291class ScanConfig: 

292 """Codebase scanning configuration.""" 

293 

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

295 exclude_patterns: list[str] = field( 

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

297 ) 

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

299 

300 @classmethod 

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

302 """Create ScanConfig from dictionary.""" 

303 return cls( 

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

305 exclude_patterns=data.get( 

306 "exclude_patterns", 

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

308 ), 

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

310 ) 

311 

312 

313@dataclass 

314class DesignTokensConfig: 

315 """Design system token configuration. 

316 

317 `active` and `profiles_dir` (ENH-1768) select one of several bundled 

318 profiles. The loader resolves token files from 

319 `<path>/<profiles_dir or "profiles">/<active>/`, with a fallback to the 

320 legacy flat layout (`<path>/primitives.json`, ...) for pre-ENH-1768 

321 projects that haven't been re-initialized. 

322 """ 

323 

324 enabled: bool = True 

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

326 primitives_file: str = "primitives.json" 

327 semantic_file: str = "semantic.json" 

328 themes_dir: str = "themes" 

329 active_theme: str = "dark" 

330 active: str = "default" 

331 profiles_dir: str | None = None 

332 

333 @classmethod 

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

335 """Create DesignTokensConfig from dictionary.""" 

336 return cls( 

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

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

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

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

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

342 active_theme=data.get("active_theme", "dark"), 

343 active=data.get("active", "default"), 

344 profiles_dir=data.get("profiles_dir"), 

345 ) 

346 

347 

348@dataclass 

349class SprintsConfig: 

350 """Sprint management configuration.""" 

351 

352 sprints_dir: str = ".sprints" 

353 default_timeout: int = 3600 

354 default_max_workers: int = 2 

355 max_issue_wall_clock_time: int = 2700 

356 

357 @classmethod 

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

359 """Create SprintsConfig from dictionary.""" 

360 return cls( 

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

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

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

364 max_issue_wall_clock_time=data.get("max_issue_wall_clock_time", 2700), 

365 ) 

366 

367 

368@dataclass 

369class DiscoverabilityConfig: 

370 """Controls how learning-test gaps are surfaced during implementation.""" 

371 

372 mode: str = "warn" 

373 skip_packages: list[str] = field(default_factory=lambda: ["std", "typing", "os", "sys"]) 

374 

375 @classmethod 

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

377 """Create DiscoverabilityConfig from dictionary.""" 

378 return cls( 

379 mode=data.get("mode", "warn"), 

380 skip_packages=data.get("skip_packages", ["std", "typing", "os", "sys"]), 

381 ) 

382 

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

384 """Serialize DiscoverabilityConfig to dictionary.""" 

385 return { 

386 "mode": self.mode, 

387 "skip_packages": list(self.skip_packages), 

388 } 

389 

390 

391@dataclass 

392class LearningTestsConfig: 

393 """Learning test registry configuration.""" 

394 

395 enabled: bool = False 

396 stale_after_days: int = 30 

397 discoverability: DiscoverabilityConfig = field(default_factory=DiscoverabilityConfig) 

398 release_gate: str = "warn" 

399 scan_dirs: list[str] = field(default_factory=lambda: ["scripts/"]) 

400 

401 @classmethod 

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

403 """Create LearningTestsConfig from dictionary.""" 

404 return cls( 

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

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

407 discoverability=DiscoverabilityConfig.from_dict(data.get("discoverability", {})), 

408 release_gate=data.get("release_gate", "warn"), 

409 scan_dirs=data.get("scan_dirs", ["scripts/"]), 

410 ) 

411 

412 

413@dataclass 

414class DecisionsConfig: 

415 """Decisions and rules log configuration.""" 

416 

417 enabled: bool = False 

418 log_path: str = ".ll/decisions.yaml" 

419 auto_generate: list[str] = field(default_factory=list) 

420 """Issue type prefixes to include during ``ll-issues decisions generate``. 

421 

422 Empty list (default) generates entries for all completed issue types. 

423 Non-empty list restricts to the listed prefixes (e.g. ``["FEAT", "ENH"]`` 

424 skips BUG entries). 

425 """ 

426 

427 @classmethod 

428 def from_dict(cls, data: dict[str, Any]) -> DecisionsConfig: 

429 """Create DecisionsConfig from dictionary.""" 

430 return cls( 

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

432 log_path=data.get("log_path", ".ll/decisions.yaml"), 

433 auto_generate=data.get("auto_generate", []), 

434 ) 

435 

436 

437@dataclass 

438class AnalyticsCaptureConfig: 

439 """Configuration for analytics capture gating (ENH-1840). 

440 

441 Controls which skills, CLI commands, and event types are captured into the 

442 unified session store. Used by ENH-1841 write-path gating via 

443 ``feature_enabled_for()``. 

444 """ 

445 

446 skills: list[str] = field(default_factory=lambda: ["*"]) 

447 cli_commands: list[str] = field(default_factory=lambda: ["*"]) 

448 corrections: bool = True 

449 file_events: bool = True 

450 correction_patterns: list[str] = field(default_factory=list) 

451 

452 @classmethod 

453 def from_dict(cls, data: dict[str, Any]) -> AnalyticsCaptureConfig: 

454 """Create AnalyticsCaptureConfig from dictionary.""" 

455 raw = data.get("correction_patterns", []) 

456 correction_patterns = ( 

457 [p for p in raw if isinstance(p, str)] if isinstance(raw, list) else [] 

458 ) 

459 return cls( 

460 skills=data.get("skills", ["*"]), 

461 cli_commands=data.get("cli_commands", ["*"]), 

462 corrections=data.get("corrections", True), 

463 file_events=data.get("file_events", True), 

464 correction_patterns=correction_patterns, 

465 ) 

466 

467 

468@dataclass 

469class LoopsGlyphsConfig: 

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

471 

472 prompt: str = "\u2726" # ✦ 

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

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

475 mcp_tool: str = "\u26a1" # ⚡ 

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

477 route: str = "\u2443" # ⑃ 

478 parallel: str = "\u2225" # ∥ 

479 

480 @classmethod 

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

482 """Create LoopsGlyphsConfig from dictionary.""" 

483 return cls( 

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

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

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

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

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

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

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

491 ) 

492 

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

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

495 return { 

496 "prompt": self.prompt, 

497 "slash_command": self.slash_command, 

498 "shell": self.shell, 

499 "mcp_tool": self.mcp_tool, 

500 "sub_loop": self.sub_loop, 

501 "route": self.route, 

502 "parallel": self.parallel, 

503 } 

504 

505 

506_VALID_SHOW_DIAGRAMS: frozenset[str] = frozenset( 

507 { 

508 "layered", 

509 "neighborhood", 

510 "inline", # topologies 

511 "detailed", 

512 "summary", 

513 "clean", 

514 "local", # presets 

515 "slim", 

516 "oneline", # presets (continued) 

517 "default", # bare --show-diagrams sentinel 

518 } 

519) 

520 

521 

522@dataclass 

523class LoopRunDefaults: 

524 """Persistent CLI defaults for ``ll-loop run``.""" 

525 

526 clear: bool = False 

527 show_diagrams: str | None = None 

528 mode: str | None = None 

529 include: str = "" 

530 

531 @classmethod 

532 def from_dict(cls, data: dict[str, Any]) -> LoopRunDefaults: 

533 """Create LoopRunDefaults from dictionary, validating show_diagrams.""" 

534 show_diagrams = data.get("show_diagrams", None) 

535 if show_diagrams is not None and show_diagrams not in _VALID_SHOW_DIAGRAMS: 

536 raise ValueError( 

537 f"loops.run_defaults.show_diagrams: {show_diagrams!r} is not valid. " 

538 f"Valid values: {sorted(_VALID_SHOW_DIAGRAMS)}" 

539 ) 

540 return cls( 

541 clear=data.get("clear", False), 

542 show_diagrams=show_diagrams, 

543 mode=data.get("mode", None), 

544 include=data.get("include", ""), 

545 ) 

546 

547 

548@dataclass 

549class LoopsConfig: 

550 """FSM loop configuration.""" 

551 

552 loops_dir: str = ".loops" 

553 queue_wait_timeout_seconds: int = 86400 

554 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

555 run_defaults: LoopRunDefaults = field(default_factory=LoopRunDefaults) 

556 

557 @classmethod 

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

559 """Create LoopsConfig from dictionary.""" 

560 return cls( 

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

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

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

564 run_defaults=LoopRunDefaults.from_dict(data.get("run_defaults", {})), 

565 ) 

566 

567 

568@dataclass 

569class GitHubSyncConfig: 

570 """GitHub-specific sync configuration.""" 

571 

572 repo: str | None = None 

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

574 default_factory=lambda: { 

575 "BUG": "bug", 

576 "FEAT": "enhancement", 

577 "ENH": "enhancement", 

578 "EPIC": "epic", 

579 } 

580 ) 

581 priority_labels: bool = True 

582 sync_completed: bool = False 

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

584 pull_template: str = "minimal" 

585 pull_limit: int = 500 

586 

587 @classmethod 

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

589 """Create GitHubSyncConfig from dictionary.""" 

590 return cls( 

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

592 label_mapping=data.get( 

593 "label_mapping", 

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

595 ), 

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

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

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

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

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

601 ) 

602 

603 

604@dataclass 

605class SyncConfig: 

606 """Issue sync configuration.""" 

607 

608 enabled: bool = False 

609 provider: str = "github" 

610 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

611 

612 @classmethod 

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

614 """Create SyncConfig from dictionary.""" 

615 return cls( 

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

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

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

619 ) 

620 

621 

622@dataclass 

623class SocketEventsConfig: 

624 """UnixSocketTransport configuration.""" 

625 

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

627 max_clients: int = 32 

628 

629 @classmethod 

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

631 """Create SocketEventsConfig from dictionary.""" 

632 return cls( 

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

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

635 ) 

636 

637 

638@dataclass 

639class OTelEventsConfig: 

640 """OTelTransport configuration.""" 

641 

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

643 service_name: str = "little-loops" 

644 

645 @classmethod 

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

647 """Create OTelEventsConfig from dictionary.""" 

648 return cls( 

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

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

651 ) 

652 

653 

654@dataclass 

655class WebhookEventsConfig: 

656 """WebhookTransport configuration.""" 

657 

658 url: str | None = None 

659 batch_ms: int = 1000 

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

661 

662 @classmethod 

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

664 """Create WebhookEventsConfig from dictionary.""" 

665 return cls( 

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

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

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

669 ) 

670 

671 

672@dataclass 

673class SqliteEventsConfig: 

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

675 

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

677 

678 @classmethod 

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

680 """Create SqliteEventsConfig from dictionary.""" 

681 return cls( 

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

683 ) 

684 

685 

686@dataclass 

687class EventsConfig: 

688 """Event transport configuration. 

689 

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

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

692 unknown names are skipped with a warning. 

693 """ 

694 

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

696 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

697 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

698 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

699 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

700 

701 @classmethod 

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

703 """Create EventsConfig from dictionary.""" 

704 return cls( 

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

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

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

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

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

710 ) 

711 

712 

713@dataclass 

714class SessionDigestConfig: 

715 """Session digest injection configuration (ENH-1907).""" 

716 

717 enabled: bool = True 

718 days: int = 7 

719 char_cap: int = 1200 

720 sections: list[str] = field(default_factory=list) 

721 

722 @classmethod 

723 def from_dict(cls, data: dict[str, Any]) -> SessionDigestConfig: 

724 """Create SessionDigestConfig from dictionary.""" 

725 return cls( 

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

727 days=data.get("days", 7), 

728 char_cap=data.get("char_cap", 1200), 

729 sections=data.get("sections", []), 

730 ) 

731 

732 

733@dataclass 

734class EvolutionConfig: 

735 """Feedback evolution configuration (ENH-1911).""" 

736 

737 feedback_min_recurrence: int = 2 

738 bypass_min_count: int = 2 

739 

740 @classmethod 

741 def from_dict(cls, data: dict[str, Any]) -> EvolutionConfig: 

742 """Create EvolutionConfig from dictionary.""" 

743 return cls( 

744 feedback_min_recurrence=data.get("feedback_min_recurrence", 2), 

745 bypass_min_count=data.get("bypass_min_count", 2), 

746 ) 

747 

748 

749@dataclass 

750class GoNoGoConfig: 

751 """Go/no-go history configuration (ENH-1914).""" 

752 

753 correction_penalty: float = -0.2 

754 

755 @classmethod 

756 def from_dict(cls, data: dict[str, Any]) -> GoNoGoConfig: 

757 """Create GoNoGoConfig from dictionary.""" 

758 return cls( 

759 correction_penalty=data.get("correction_penalty", -0.2), 

760 ) 

761 

762 

763@dataclass 

764class CaptureIssueConfig: 

765 """Capture-issue history configuration (ENH-1914).""" 

766 

767 dup_overlap_threshold: float = 0.7 

768 

769 @classmethod 

770 def from_dict(cls, data: dict[str, Any]) -> CaptureIssueConfig: 

771 """Create CaptureIssueConfig from dictionary.""" 

772 return cls( 

773 dup_overlap_threshold=data.get("dup_overlap_threshold", 0.7), 

774 ) 

775 

776 

777@dataclass 

778class CompactionConfig: 

779 """LCM-style compaction configuration for summary_nodes (FEAT-1712). 

780 

781 Controls whether backfill() generates LLM summaries over message_events blocks 

782 via three-level LCM Algorithm 3 escalation (normal → aggressive bullet-point → 

783 deterministic truncation). Disabled by default to avoid background LLM calls 

784 without user opt-in. 

785 

786 Cross-session condensation (ENH-1954): when ``cross_session_enabled`` is True 

787 (default), the compaction pass recurses over existing condensed nodes level by 

788 level, grouping by token budget and summarising each group until exactly one 

789 project-root summary node remains (``session_id IS NULL``, ``level = max``). 

790 """ 

791 

792 enabled: bool = False 

793 budget_tokens: int = 4096 

794 model: str | None = None 

795 timeout: int = 60 

796 cross_session_enabled: bool = True 

797 max_level: int | None = None 

798 

799 @classmethod 

800 def from_dict(cls, data: dict[str, Any]) -> CompactionConfig: 

801 """Create CompactionConfig from dictionary. Lenient: ignores unknown keys.""" 

802 return cls( 

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

804 budget_tokens=data.get("budget_tokens", 4096), 

805 model=data.get("model", None), 

806 timeout=data.get("timeout", 60), 

807 cross_session_enabled=data.get("cross_session_enabled", True), 

808 max_level=data.get("max_level", None), 

809 ) 

810 

811 

812@dataclass 

813class RetentionConfig: 

814 """Retention policy for history.db raw event tables (ENH-1906). 

815 

816 Pruning is dual-gated: both ``min_project_age_days`` and ``min_db_size_mb`` 

817 must be exceeded before any rows are deleted. Default thresholds are generous 

818 so fresh or small projects are never affected. 

819 

820 Only high-volume tables are pruned (tool_events, cli_events, file_events, 

821 message_events). High-value tables (issue_events, user_corrections) are 

822 never pruned. 

823 """ 

824 

825 min_project_age_days: int = 365 

826 min_db_size_mb: int = 800 

827 raw_event_max_age_days: int | None = 90 

828 

829 @classmethod 

830 def from_dict(cls, data: dict[str, Any]) -> RetentionConfig: 

831 """Create RetentionConfig from dictionary. Lenient: ignores unknown keys.""" 

832 return cls( 

833 min_project_age_days=data.get("min_project_age_days", 365), 

834 min_db_size_mb=data.get("min_db_size_mb", 800),