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

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 

530 @classmethod 

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

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

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

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

535 raise ValueError( 

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

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

538 ) 

539 return cls( 

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

541 show_diagrams=show_diagrams, 

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

543 ) 

544 

545 

546@dataclass 

547class LoopsConfig: 

548 """FSM loop configuration.""" 

549 

550 loops_dir: str = ".loops" 

551 queue_wait_timeout_seconds: int = 86400 

552 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

553 run_defaults: LoopRunDefaults = field(default_factory=LoopRunDefaults) 

554 

555 @classmethod 

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

557 """Create LoopsConfig from dictionary.""" 

558 return cls( 

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

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

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

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

563 ) 

564 

565 

566@dataclass 

567class GitHubSyncConfig: 

568 """GitHub-specific sync configuration.""" 

569 

570 repo: str | None = None 

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

572 default_factory=lambda: { 

573 "BUG": "bug", 

574 "FEAT": "enhancement", 

575 "ENH": "enhancement", 

576 "EPIC": "epic", 

577 } 

578 ) 

579 priority_labels: bool = True 

580 sync_completed: bool = False 

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

582 pull_template: str = "minimal" 

583 pull_limit: int = 500 

584 

585 @classmethod 

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

587 """Create GitHubSyncConfig from dictionary.""" 

588 return cls( 

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

590 label_mapping=data.get( 

591 "label_mapping", 

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

593 ), 

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

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

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

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

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

599 ) 

600 

601 

602@dataclass 

603class SyncConfig: 

604 """Issue sync configuration.""" 

605 

606 enabled: bool = False 

607 provider: str = "github" 

608 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

609 

610 @classmethod 

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

612 """Create SyncConfig from dictionary.""" 

613 return cls( 

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

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

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

617 ) 

618 

619 

620@dataclass 

621class SocketEventsConfig: 

622 """UnixSocketTransport configuration.""" 

623 

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

625 max_clients: int = 32 

626 

627 @classmethod 

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

629 """Create SocketEventsConfig from dictionary.""" 

630 return cls( 

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

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

633 ) 

634 

635 

636@dataclass 

637class OTelEventsConfig: 

638 """OTelTransport configuration.""" 

639 

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

641 service_name: str = "little-loops" 

642 

643 @classmethod 

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

645 """Create OTelEventsConfig from dictionary.""" 

646 return cls( 

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

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

649 ) 

650 

651 

652@dataclass 

653class WebhookEventsConfig: 

654 """WebhookTransport configuration.""" 

655 

656 url: str | None = None 

657 batch_ms: int = 1000 

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

659 

660 @classmethod 

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

662 """Create WebhookEventsConfig from dictionary.""" 

663 return cls( 

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

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

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

667 ) 

668 

669 

670@dataclass 

671class SqliteEventsConfig: 

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

673 

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

675 

676 @classmethod 

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

678 """Create SqliteEventsConfig from dictionary.""" 

679 return cls( 

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

681 ) 

682 

683 

684@dataclass 

685class EventsConfig: 

686 """Event transport configuration. 

687 

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

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

690 unknown names are skipped with a warning. 

691 """ 

692 

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

694 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

695 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

696 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

697 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

698 

699 @classmethod 

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

701 """Create EventsConfig from dictionary.""" 

702 return cls( 

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

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

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

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

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

708 ) 

709 

710 

711@dataclass 

712class SessionDigestConfig: 

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

714 

715 enabled: bool = True 

716 days: int = 7 

717 char_cap: int = 1200 

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

719 

720 @classmethod 

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

722 """Create SessionDigestConfig from dictionary.""" 

723 return cls( 

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

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

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

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

728 ) 

729 

730 

731@dataclass 

732class EvolutionConfig: 

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

734 

735 feedback_min_recurrence: int = 2 

736 bypass_min_count: int = 2 

737 

738 @classmethod 

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

740 """Create EvolutionConfig from dictionary.""" 

741 return cls( 

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

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

744 ) 

745 

746 

747@dataclass 

748class GoNoGoConfig: 

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

750 

751 correction_penalty: float = -0.2 

752 

753 @classmethod 

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

755 """Create GoNoGoConfig from dictionary.""" 

756 return cls( 

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

758 ) 

759 

760 

761@dataclass 

762class CaptureIssueConfig: 

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

764 

765 dup_overlap_threshold: float = 0.7 

766 

767 @classmethod 

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

769 """Create CaptureIssueConfig from dictionary.""" 

770 return cls( 

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

772 ) 

773 

774 

775@dataclass 

776class CompactionConfig: 

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

778 

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

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

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

782 without user opt-in. 

783 

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

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

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

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

788 """ 

789 

790 enabled: bool = False 

791 budget_tokens: int = 4096 

792 model: str | None = None 

793 timeout: int = 60 

794 cross_session_enabled: bool = True 

795 max_level: int | None = None 

796 

797 @classmethod 

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

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

800 return cls( 

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

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

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

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

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

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

807 ) 

808 

809 

810@dataclass 

811class RetentionConfig: 

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

813 

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

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

816 so fresh or small projects are never affected. 

817 

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

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

820 never pruned. 

821 """ 

822 

823 min_project_age_days: int = 365 

824 min_db_size_mb: int = 800 

825 raw_event_max_age_days: int | None = 90 

826 

827 @classmethod 

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

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

830 return cls( 

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

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

833 raw_event_max_age_days=data.get("raw_event_max_age_days", 90), 

834 )