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 

356 @classmethod 

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

358 """Create SprintsConfig from dictionary.""" 

359 return cls( 

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

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

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

363 ) 

364 

365 

366@dataclass 

367class DiscoverabilityConfig: 

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

369 

370 mode: str = "warn" 

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

372 

373 @classmethod 

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

375 """Create DiscoverabilityConfig from dictionary.""" 

376 return cls( 

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

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

379 ) 

380 

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

382 """Serialize DiscoverabilityConfig to dictionary.""" 

383 return { 

384 "mode": self.mode, 

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

386 } 

387 

388 

389@dataclass 

390class LearningTestsConfig: 

391 """Learning test registry configuration.""" 

392 

393 enabled: bool = False 

394 stale_after_days: int = 30 

395 discoverability: DiscoverabilityConfig = field(default_factory=DiscoverabilityConfig) 

396 

397 @classmethod 

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

399 """Create LearningTestsConfig from dictionary.""" 

400 return cls( 

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

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

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

404 ) 

405 

406 

407@dataclass 

408class DecisionsConfig: 

409 """Decisions and rules log configuration.""" 

410 

411 enabled: bool = False 

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

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

414 

415 @classmethod 

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

417 """Create DecisionsConfig from dictionary.""" 

418 return cls( 

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

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

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

422 ) 

423 

424 

425@dataclass 

426class AnalyticsCaptureConfig: 

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

428 

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

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

431 ``feature_enabled_for()``. 

432 """ 

433 

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

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

436 corrections: bool = True 

437 file_events: bool = True 

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

439 

440 @classmethod 

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

442 """Create AnalyticsCaptureConfig from dictionary.""" 

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

444 correction_patterns = ( 

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

446 ) 

447 return cls( 

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

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

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

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

452 correction_patterns=correction_patterns, 

453 ) 

454 

455 

456@dataclass 

457class LoopsGlyphsConfig: 

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

459 

460 prompt: str = "\u2726" # ✦ 

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

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

463 mcp_tool: str = "\u26a1" # ⚡ 

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

465 route: str = "\u2443" # ⑃ 

466 parallel: str = "\u2225" # ∥ 

467 

468 @classmethod 

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

470 """Create LoopsGlyphsConfig from dictionary.""" 

471 return cls( 

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

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

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

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

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

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

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

479 ) 

480 

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

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

483 return { 

484 "prompt": self.prompt, 

485 "slash_command": self.slash_command, 

486 "shell": self.shell, 

487 "mcp_tool": self.mcp_tool, 

488 "sub_loop": self.sub_loop, 

489 "route": self.route, 

490 "parallel": self.parallel, 

491 } 

492 

493 

494@dataclass 

495class LoopsConfig: 

496 """FSM loop configuration.""" 

497 

498 loops_dir: str = ".loops" 

499 queue_wait_timeout_seconds: int = 86400 

500 glyphs: LoopsGlyphsConfig = field(default_factory=LoopsGlyphsConfig) 

501 

502 @classmethod 

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

504 """Create LoopsConfig from dictionary.""" 

505 return cls( 

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

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

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

509 ) 

510 

511 

512@dataclass 

513class GitHubSyncConfig: 

514 """GitHub-specific sync configuration.""" 

515 

516 repo: str | None = None 

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

518 default_factory=lambda: { 

519 "BUG": "bug", 

520 "FEAT": "enhancement", 

521 "ENH": "enhancement", 

522 "EPIC": "epic", 

523 } 

524 ) 

525 priority_labels: bool = True 

526 sync_completed: bool = False 

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

528 pull_template: str = "minimal" 

529 pull_limit: int = 500 

530 

531 @classmethod 

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

533 """Create GitHubSyncConfig from dictionary.""" 

534 return cls( 

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

536 label_mapping=data.get( 

537 "label_mapping", 

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

539 ), 

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

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

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

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

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

545 ) 

546 

547 

548@dataclass 

549class SyncConfig: 

550 """Issue sync configuration.""" 

551 

552 enabled: bool = False 

553 provider: str = "github" 

554 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

555 

556 @classmethod 

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

558 """Create SyncConfig from dictionary.""" 

559 return cls( 

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

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

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

563 ) 

564 

565 

566@dataclass 

567class SocketEventsConfig: 

568 """UnixSocketTransport configuration.""" 

569 

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

571 max_clients: int = 32 

572 

573 @classmethod 

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

575 """Create SocketEventsConfig from dictionary.""" 

576 return cls( 

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

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

579 ) 

580 

581 

582@dataclass 

583class OTelEventsConfig: 

584 """OTelTransport configuration.""" 

585 

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

587 service_name: str = "little-loops" 

588 

589 @classmethod 

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

591 """Create OTelEventsConfig from dictionary.""" 

592 return cls( 

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

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

595 ) 

596 

597 

598@dataclass 

599class WebhookEventsConfig: 

600 """WebhookTransport configuration.""" 

601 

602 url: str | None = None 

603 batch_ms: int = 1000 

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

605 

606 @classmethod 

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

608 """Create WebhookEventsConfig from dictionary.""" 

609 return cls( 

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

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

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

613 ) 

614 

615 

616@dataclass 

617class SqliteEventsConfig: 

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

619 

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

621 

622 @classmethod 

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

624 """Create SqliteEventsConfig from dictionary.""" 

625 return cls( 

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

627 ) 

628 

629 

630@dataclass 

631class EventsConfig: 

632 """Event transport configuration. 

633 

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

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

636 unknown names are skipped with a warning. 

637 """ 

638 

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

640 socket: SocketEventsConfig = field(default_factory=SocketEventsConfig) 

641 otel: OTelEventsConfig = field(default_factory=OTelEventsConfig) 

642 webhook: WebhookEventsConfig = field(default_factory=WebhookEventsConfig) 

643 sqlite: SqliteEventsConfig = field(default_factory=SqliteEventsConfig) 

644 

645 @classmethod 

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

647 """Create EventsConfig from dictionary.""" 

648 return cls( 

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

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

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

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

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

654 ) 

655 

656 

657@dataclass 

658class SessionDigestConfig: 

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

660 

661 enabled: bool = False 

662 days: int = 7 

663 char_cap: int = 1200 

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

665 

666 @classmethod 

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

668 """Create SessionDigestConfig from dictionary.""" 

669 return cls( 

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

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

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

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

674 ) 

675 

676 

677@dataclass 

678class EvolutionConfig: 

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

680 

681 feedback_min_recurrence: int = 2 

682 bypass_min_count: int = 2 

683 

684 @classmethod 

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

686 """Create EvolutionConfig from dictionary.""" 

687 return cls( 

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

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

690 ) 

691 

692 

693@dataclass 

694class GoNoGoConfig: 

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

696 

697 correction_penalty: float = -0.2 

698 

699 @classmethod 

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

701 """Create GoNoGoConfig from dictionary.""" 

702 return cls( 

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

704 ) 

705 

706 

707@dataclass 

708class CaptureIssueConfig: 

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

710 

711 dup_overlap_threshold: float = 0.7 

712 

713 @classmethod 

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

715 """Create CaptureIssueConfig from dictionary.""" 

716 return cls( 

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

718 ) 

719 

720 

721@dataclass 

722class CompactionConfig: 

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

724 

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

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

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

728 without user opt-in. 

729 

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

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

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

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

734 """ 

735 

736 enabled: bool = False 

737 budget_tokens: int = 4096 

738 model: str | None = None 

739 timeout: int = 60 

740 cross_session_enabled: bool = True 

741 max_level: int | None = None 

742 

743 @classmethod 

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

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

746 return cls( 

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

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

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

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

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

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

753 ) 

754 

755 

756@dataclass 

757class RetentionConfig: 

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

759 

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

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

762 so fresh or small projects are never affected. 

763 

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

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

766 never pruned. 

767 """ 

768 

769 min_project_age_days: int = 365 

770 min_db_size_mb: int = 800 

771 raw_event_max_age_days: int | None = 90 

772 

773 @classmethod 

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

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

776 return cls( 

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

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

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

780 ) 

781 

782 

783@dataclass 

784class HistoryConfig: 

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

786 

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

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

789 """ 

790 

791 velocity_window: int = 10 

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

793 max_age_days: int | None = None 

794 planning_skills: list[str] = field( 

795 default_factory=lambda: ["create-sprint", "scope-epic", "manage-issue", "review-epic"] 

796 ) 

797 session_digest: SessionDigestConfig = field(default_factory=SessionDigestConfig) 

798 evolution: EvolutionConfig = field(default_factory=EvolutionConfig) 

799 go_no_go: GoNoGoConfig = field(default_factory=GoNoGoConfig) 

800 capture_issue: CaptureIssueConfig = field(default_factory=CaptureIssueConfig) 

801 compaction: CompactionConfig = field(default_factory=CompactionConfig) 

802 

803 @classmethod 

804 def from_dict(cls, data: dict[str, Any]) -> HistoryConfig: 

805 """Create HistoryConfig from dictionary. Lenient: ignores unknown keys, never raises.""" 

806 return cls( 

807 velocity_window=data.get("velocity_window", 10), 

808 effort_fields=data.get("effort_fields", ["session_count", "cycle_time_days"]), 

809 max_age_days=data.get("max_age_days", None), 

810 planning_skills=data.get( 

811 "planning_skills", 

812 ["create-sprint", "scope-epic", "manage-issue", "review-epic"], 

813 ), 

814 session_digest=SessionDigestConfig.from_dict(data.get("session_digest", {})), 

815 evolution=EvolutionConfig.from_dict(data.get("evolution", {})), 

816 go_no_go=GoNoGoConfig.from_dict(data.get("go_no_go", {})), 

817 capture_issue=CaptureIssueConfig.from_dict(data.get("capture_issue", {})), 

818 compaction=CompactionConfig.from_dict(data.get("compaction", {})), 

819 )