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 

399 @classmethod 

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

401 """Create LearningTestsConfig from dictionary.""" 

402 return cls( 

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

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

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

406 ) 

407 

408 

409@dataclass 

410class DecisionsConfig: 

411 """Decisions and rules log configuration.""" 

412 

413 enabled: bool = False 

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

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

416 

417 @classmethod 

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

419 """Create DecisionsConfig from dictionary.""" 

420 return cls( 

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

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

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

424 ) 

425 

426 

427@dataclass 

428class AnalyticsCaptureConfig: 

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

430 

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

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

433 ``feature_enabled_for()``. 

434 """ 

435 

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

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

438 corrections: bool = True 

439 file_events: bool = True 

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

441 

442 @classmethod 

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

444 """Create AnalyticsCaptureConfig from dictionary.""" 

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

446 correction_patterns = ( 

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

448 ) 

449 return cls( 

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

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

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

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

454 correction_patterns=correction_patterns, 

455 ) 

456 

457 

458@dataclass 

459class LoopsGlyphsConfig: 

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

461 

462 prompt: str = "\u2726" # ✦ 

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

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

465 mcp_tool: str = "\u26a1" # ⚡ 

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

467 route: str = "\u2443" # ⑃ 

468 parallel: str = "\u2225" # ∥ 

469 

470 @classmethod 

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

472 """Create LoopsGlyphsConfig from dictionary.""" 

473 return cls( 

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

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

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

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

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

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

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

481 ) 

482 

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

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

485 return { 

486 "prompt": self.prompt, 

487 "slash_command": self.slash_command, 

488 "shell": self.shell, 

489 "mcp_tool": self.mcp_tool, 

490 "sub_loop": self.sub_loop, 

491 "route": self.route, 

492 "parallel": self.parallel, 

493 } 

494 

495 

496_VALID_SHOW_DIAGRAMS: frozenset[str] = frozenset( 

497 { 

498 "layered", 

499 "neighborhood", 

500 "inline", # topologies 

501 "detailed", 

502 "summary", 

503 "clean", 

504 "local", # presets 

505 "slim", 

506 "oneline", # presets (continued) 

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

508 } 

509) 

510 

511 

512@dataclass 

513class LoopRunDefaults: 

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

515 

516 clear: bool = False 

517 show_diagrams: str | None = None 

518 mode: str | None = None 

519 

520 @classmethod 

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

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

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

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

525 raise ValueError( 

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

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

528 ) 

529 return cls( 

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

531 show_diagrams=show_diagrams, 

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

533 ) 

534 

535 

536@dataclass 

537class LoopsConfig: 

538 """FSM loop configuration.""" 

539 

540 loops_dir: str = ".loops" 

541 queue_wait_timeout_seconds: int = 86400 

542 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

543 run_defaults: LoopRunDefaults = field(default_factory=LoopRunDefaults) 

544 

545 @classmethod 

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

547 """Create LoopsConfig from dictionary.""" 

548 return cls( 

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

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

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

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

553 ) 

554 

555 

556@dataclass 

557class GitHubSyncConfig: 

558 """GitHub-specific sync configuration.""" 

559 

560 repo: str | None = None 

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

562 default_factory=lambda: { 

563 "BUG": "bug", 

564 "FEAT": "enhancement", 

565 "ENH": "enhancement", 

566 "EPIC": "epic", 

567 } 

568 ) 

569 priority_labels: bool = True 

570 sync_completed: bool = False 

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

572 pull_template: str = "minimal" 

573 pull_limit: int = 500 

574 

575 @classmethod 

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

577 """Create GitHubSyncConfig from dictionary.""" 

578 return cls( 

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

580 label_mapping=data.get( 

581 "label_mapping", 

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

583 ), 

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

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

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

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

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

589 ) 

590 

591 

592@dataclass 

593class SyncConfig: 

594 """Issue sync configuration.""" 

595 

596 enabled: bool = False 

597 provider: str = "github" 

598 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

599 

600 @classmethod 

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

602 """Create SyncConfig from dictionary.""" 

603 return cls( 

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

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

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

607 ) 

608 

609 

610@dataclass 

611class SocketEventsConfig: 

612 """UnixSocketTransport configuration.""" 

613 

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

615 max_clients: int = 32 

616 

617 @classmethod 

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

619 """Create SocketEventsConfig from dictionary.""" 

620 return cls( 

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

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

623 ) 

624 

625 

626@dataclass 

627class OTelEventsConfig: 

628 """OTelTransport configuration.""" 

629 

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

631 service_name: str = "little-loops" 

632 

633 @classmethod 

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

635 """Create OTelEventsConfig from dictionary.""" 

636 return cls( 

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

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

639 ) 

640 

641 

642@dataclass 

643class WebhookEventsConfig: 

644 """WebhookTransport configuration.""" 

645 

646 url: str | None = None 

647 batch_ms: int = 1000 

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

649 

650 @classmethod 

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

652 """Create WebhookEventsConfig from dictionary.""" 

653 return cls( 

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

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

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

657 ) 

658 

659 

660@dataclass 

661class SqliteEventsConfig: 

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

663 

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

665 

666 @classmethod 

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

668 """Create SqliteEventsConfig from dictionary.""" 

669 return cls( 

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

671 ) 

672 

673 

674@dataclass 

675class EventsConfig: 

676 """Event transport configuration. 

677 

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

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

680 unknown names are skipped with a warning. 

681 """ 

682 

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

684 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

685 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

686 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

687 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

688 

689 @classmethod 

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

691 """Create EventsConfig from dictionary.""" 

692 return cls( 

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

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

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

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

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

698 ) 

699 

700 

701@dataclass 

702class SessionDigestConfig: 

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

704 

705 enabled: bool = True 

706 days: int = 7 

707 char_cap: int = 800 

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

709 

710 @classmethod 

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

712 """Create SessionDigestConfig from dictionary.""" 

713 return cls( 

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

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

716 char_cap=data.get("char_cap", 800), 

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

718 ) 

719 

720 

721@dataclass 

722class EvolutionConfig: 

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

724 

725 feedback_min_recurrence: int = 2 

726 bypass_min_count: int = 2 

727 

728 @classmethod 

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

730 """Create EvolutionConfig from dictionary.""" 

731 return cls( 

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

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

734 ) 

735 

736 

737@dataclass 

738class GoNoGoConfig: 

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

740 

741 correction_penalty: float = -0.2 

742 

743 @classmethod 

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

745 """Create GoNoGoConfig from dictionary.""" 

746 return cls( 

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

748 ) 

749 

750 

751@dataclass 

752class CaptureIssueConfig: 

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

754 

755 dup_overlap_threshold: float = 0.7 

756 

757 @classmethod 

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

759 """Create CaptureIssueConfig from dictionary.""" 

760 return cls( 

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

762 ) 

763 

764 

765@dataclass 

766class CompactionConfig: 

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

768 

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

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

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

772 without user opt-in. 

773 

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

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

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

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

778 """ 

779 

780 enabled: bool = False 

781 budget_tokens: int = 4096 

782 model: str | None = None 

783 timeout: int = 60 

784 cross_session_enabled: bool = True 

785 max_level: int | None = None 

786 

787 @classmethod 

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

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

790 return cls( 

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

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

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

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

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

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

797 ) 

798 

799 

800@dataclass 

801class RetentionConfig: 

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

803 

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

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

806 so fresh or small projects are never affected. 

807 

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

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

810 never pruned. 

811 """ 

812 

813 min_project_age_days: int = 365 

814 min_db_size_mb: int = 800 

815 raw_event_max_age_days: int | None = 90 

816 

817 @classmethod 

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

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

820 return cls( 

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

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

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

824 ) 

825 

826 

827@dataclass 

828class HistoryConfig: 

829 """History read/consume configuration (ENH-1913). 

830 

831 Single, consistent foundation for .ll/history.db read/consume configurability. 

832 Owns the complete history.* namespace; consumers wire runtime + CLI only. 

833 """ 

834 

835 velocity_window: int = 10 

836 effort_fields: list[str] = field(default_factory=lambda: ["session_count", "cycle_time_days"]) 

837 max_age_days: int | None = None