Coverage for little_loops / config.py: 98%

274 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2026-03-04 01:17 -0600

1"""Configuration management for little-loops. 

2 

3Provides the BRConfig class for loading, merging, and accessing project configuration. 

4Configuration is read from .claude/ll-config.json and merged with sensible defaults. 

5""" 

6 

7from __future__ import annotations 

8 

9import json 

10from dataclasses import dataclass, field 

11from pathlib import Path 

12from typing import TYPE_CHECKING, Any, cast 

13 

14if TYPE_CHECKING: 

15 from little_loops.parallel.types import ParallelConfig 

16 

17__all__ = [ 

18 "BRConfig", 

19 "CLConfig", 

20 "CategoryConfig", 

21 "ProjectConfig", 

22 "IssuesConfig", 

23 "AutomationConfig", 

24 "ParallelAutomationConfig", 

25 "CommandsConfig", 

26 "ScanConfig", 

27 "SprintsConfig", 

28 "LoopsConfig", 

29 "GitHubSyncConfig", 

30 "ConfidenceGateConfig", 

31 "SyncConfig", 

32 "ScoringWeightsConfig", 

33 "DependencyMappingConfig", 

34 "REQUIRED_CATEGORIES", 

35 "DEFAULT_CATEGORIES", 

36] 

37 

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

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

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

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

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

43} 

44 

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

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

47 **REQUIRED_CATEGORIES, 

48} 

49 

50 

51@dataclass 

52class CategoryConfig: 

53 """Configuration for an issue category.""" 

54 

55 prefix: str 

56 dir: str 

57 action: str = "fix" 

58 

59 @classmethod 

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

61 """Create CategoryConfig from dictionary.""" 

62 return cls( 

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

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

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

66 ) 

67 

68 

69@dataclass 

70class ProjectConfig: 

71 """Project-level configuration.""" 

72 

73 name: str = "" 

74 src_dir: str = "src/" 

75 test_cmd: str = "pytest" 

76 lint_cmd: str = "ruff check ." 

77 type_cmd: str | None = "mypy" 

78 format_cmd: str | None = "ruff format ." 

79 build_cmd: str | None = None 

80 run_cmd: str | None = None 

81 

82 @classmethod 

83 def from_dict(cls, data: dict[str, Any]) -> ProjectConfig: 

84 """Create ProjectConfig from dictionary.""" 

85 return cls( 

86 name=data.get("name", ""), 

87 src_dir=data.get("src_dir", "src/"), 

88 test_cmd=data.get("test_cmd", "pytest"), 

89 lint_cmd=data.get("lint_cmd", "ruff check ."), 

90 type_cmd=data.get("type_cmd", "mypy"), 

91 format_cmd=data.get("format_cmd", "ruff format ."), 

92 build_cmd=data.get("build_cmd"), 

93 run_cmd=data.get("run_cmd"), 

94 ) 

95 

96 

97@dataclass 

98class IssuesConfig: 

99 """Issue management configuration.""" 

100 

101 base_dir: str = ".issues" 

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

103 completed_dir: str = "completed" 

104 deferred_dir: str = "deferred" 

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

106 templates_dir: str | None = None 

107 

108 @classmethod 

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

110 """Create IssuesConfig from dictionary. 

111 

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

113 included if not specified in user config. 

114 """ 

115 # Start with user categories or empty dict 

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

117 

118 # Ensure required categories exist (merge with defaults) 

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

120 if key not in categories_data: 

121 categories_data[key] = defaults 

122 

123 categories = { 

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

125 } 

126 return cls( 

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

128 categories=categories, 

129 completed_dir=data.get("completed_dir", "completed"), 

130 deferred_dir=data.get("deferred_dir", "deferred"), 

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

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

133 ) 

134 

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

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

137 

138 Args: 

139 prefix: Issue type prefix to look up 

140 

141 Returns: 

142 CategoryConfig if found, None otherwise 

143 """ 

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

145 if category.prefix == prefix: 

146 return category 

147 return None 

148 

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

150 """Get category config by directory name. 

151 

152 Args: 

153 dir_name: Directory name to look up 

154 

155 Returns: 

156 CategoryConfig if found, None otherwise 

157 """ 

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

159 if category.dir == dir_name: 

160 return category 

161 return None 

162 

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

164 """Get all configured issue type prefixes. 

165 

166 Returns: 

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

168 """ 

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

170 

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

172 """Get all configured issue directory names. 

173 

174 Returns: 

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

176 """ 

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

178 

179 

180@dataclass 

181class AutomationConfig: 

182 """Automation script configuration.""" 

183 

184 timeout_seconds: int = 3600 

185 idle_timeout_seconds: int = 0 # Kill if no output for N seconds (0 to disable) 

186 state_file: str = ".auto-manage-state.json" 

187 worktree_base: str = ".worktrees" 

188 max_workers: int = 2 

189 stream_output: bool = True 

190 max_continuations: int = 3 # Max session restarts on context handoff 

191 

192 @classmethod 

193 def from_dict(cls, data: dict[str, Any]) -> AutomationConfig: 

194 """Create AutomationConfig from dictionary.""" 

195 return cls( 

196 timeout_seconds=data.get("timeout_seconds", 3600), 

197 idle_timeout_seconds=data.get("idle_timeout_seconds", 0), 

198 state_file=data.get("state_file", ".auto-manage-state.json"), 

199 worktree_base=data.get("worktree_base", ".worktrees"), 

200 max_workers=data.get("max_workers", 2), 

201 stream_output=data.get("stream_output", True), 

202 max_continuations=data.get("max_continuations", 3), 

203 ) 

204 

205 

206@dataclass 

207class ParallelAutomationConfig: 

208 """Parallel automation configuration using composition. 

209 

210 Uses AutomationConfig for shared settings (max_workers, worktree_base, 

211 state_file, timeout_seconds, stream_output) plus parallel-specific fields. 

212 """ 

213 

214 base: AutomationConfig 

215 p0_sequential: bool = True 

216 max_merge_retries: int = 2 

217 command_prefix: str = "/ll:" 

218 ready_command: str = "ready-issue {{issue_id}}" 

219 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}" 

220 worktree_copy_files: list[str] = field( 

221 default_factory=lambda: [".claude/settings.local.json", ".env"] 

222 ) 

223 require_code_changes: bool = True 

224 

225 @classmethod 

226 def from_dict(cls, data: dict[str, Any]) -> ParallelAutomationConfig: 

227 """Create ParallelAutomationConfig from dictionary. 

228 

229 Shared fields use parallel-specific defaults: 

230 - state_file: ".parallel-manage-state.json" 

231 - stream_output: False 

232 """ 

233 base = AutomationConfig( 

234 timeout_seconds=data.get("timeout_seconds", 3600), 

235 state_file=data.get("state_file", ".parallel-manage-state.json"), 

236 worktree_base=data.get("worktree_base", ".worktrees"), 

237 max_workers=data.get("max_workers", 2), 

238 stream_output=data.get("stream_output", False), 

239 ) 

240 return cls( 

241 base=base, 

242 p0_sequential=data.get("p0_sequential", True), 

243 max_merge_retries=data.get("max_merge_retries", 2), 

244 command_prefix=data.get("command_prefix", "/ll:"), 

245 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"), 

246 manage_command=data.get( 

247 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}" 

248 ), 

249 worktree_copy_files=data.get( 

250 "worktree_copy_files", [".claude/settings.local.json", ".env"] 

251 ), 

252 require_code_changes=data.get("require_code_changes", True), 

253 ) 

254 

255 

256@dataclass 

257class ConfidenceGateConfig: 

258 """Confidence score gate configuration for manage-issue.""" 

259 

260 enabled: bool = False 

261 threshold: int = 85 

262 

263 @classmethod 

264 def from_dict(cls, data: dict[str, Any]) -> ConfidenceGateConfig: 

265 """Create ConfidenceGateConfig from dictionary.""" 

266 return cls( 

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

268 threshold=data.get("threshold", 85), 

269 ) 

270 

271 

272@dataclass 

273class CommandsConfig: 

274 """Command customization configuration.""" 

275 

276 pre_implement: str | None = None 

277 post_implement: str | None = None 

278 custom_verification: list[str] = field(default_factory=list) 

279 confidence_gate: ConfidenceGateConfig = field(default_factory=ConfidenceGateConfig) 

280 tdd_mode: bool = False 

281 

282 @classmethod 

283 def from_dict(cls, data: dict[str, Any]) -> CommandsConfig: 

284 """Create CommandsConfig from dictionary.""" 

285 return cls( 

286 pre_implement=data.get("pre_implement"), 

287 post_implement=data.get("post_implement"), 

288 custom_verification=data.get("custom_verification", []), 

289 confidence_gate=ConfidenceGateConfig.from_dict(data.get("confidence_gate", {})), 

290 tdd_mode=data.get("tdd_mode", False), 

291 ) 

292 

293 

294@dataclass 

295class ScanConfig: 

296 """Codebase scanning configuration.""" 

297 

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

299 exclude_patterns: list[str] = field( 

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

301 ) 

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

303 

304 @classmethod 

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

306 """Create ScanConfig from dictionary.""" 

307 return cls( 

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

309 exclude_patterns=data.get( 

310 "exclude_patterns", 

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

312 ), 

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

314 ) 

315 

316 

317@dataclass 

318class SprintsConfig: 

319 """Sprint management configuration.""" 

320 

321 sprints_dir: str = ".sprints" 

322 default_timeout: int = 3600 

323 default_max_workers: int = 2 

324 

325 @classmethod 

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

327 """Create SprintsConfig from dictionary.""" 

328 return cls( 

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

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

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

332 ) 

333 

334 

335@dataclass 

336class LoopsConfig: 

337 """FSM loop configuration.""" 

338 

339 loops_dir: str = ".loops" 

340 

341 @classmethod 

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

343 """Create LoopsConfig from dictionary.""" 

344 return cls( 

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

346 ) 

347 

348 

349@dataclass 

350class GitHubSyncConfig: 

351 """GitHub-specific sync configuration.""" 

352 

353 repo: str | None = None 

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

355 default_factory=lambda: {"BUG": "bug", "FEAT": "enhancement", "ENH": "enhancement"} 

356 ) 

357 priority_labels: bool = True 

358 sync_completed: bool = False 

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

360 pull_template: str = "minimal" 

361 

362 @classmethod 

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

364 """Create GitHubSyncConfig from dictionary.""" 

365 return cls( 

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

367 label_mapping=data.get( 

368 "label_mapping", {"BUG": "bug", "FEAT": "enhancement", "ENH": "enhancement"} 

369 ), 

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

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

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

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

374 ) 

375 

376 

377@dataclass 

378class SyncConfig: 

379 """Issue sync configuration.""" 

380 

381 enabled: bool = False 

382 provider: str = "github" 

383 github: GitHubSyncConfig = field(default_factory=GitHubSyncConfig) 

384 

385 @classmethod 

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

387 """Create SyncConfig from dictionary.""" 

388 return cls( 

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

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

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

392 ) 

393 

394 

395@dataclass 

396class ScoringWeightsConfig: 

397 """Scoring weights for semantic conflict analysis. 

398 

399 Weights for the three signals used in compute_conflict_score(). 

400 Should sum to 1.0 for normalized scoring. 

401 

402 Attributes: 

403 semantic: Weight for semantic target overlap (component/function names) 

404 section: Weight for section mention overlap (UI regions) 

405 type: Weight for modification type match 

406 """ 

407 

408 semantic: float = 0.5 

409 section: float = 0.3 

410 type: float = 0.2 

411 

412 @classmethod 

413 def from_dict(cls, data: dict[str, Any]) -> ScoringWeightsConfig: 

414 """Create ScoringWeightsConfig from dictionary.""" 

415 return cls( 

416 semantic=data.get("semantic", 0.5), 

417 section=data.get("section", 0.3), 

418 type=data.get("type", 0.2), 

419 ) 

420 

421 

422@dataclass 

423class DependencyMappingConfig: 

424 """Dependency mapping threshold configuration. 

425 

426 Controls overlap detection sensitivity and conflict scoring thresholds. 

427 Default values match the previously hardcoded constants for backwards 

428 compatibility. 

429 

430 Attributes: 

431 overlap_min_files: Minimum overlapping files to trigger overlap 

432 overlap_min_ratio: Minimum ratio of overlapping files to smaller set 

433 min_directory_depth: Minimum path segments for directory overlap 

434 conflict_threshold: Below = parallel-safe, above = dependency proposed 

435 high_conflict_threshold: Above = HIGH conflict label 

436 confidence_modifier: Applied when dependency direction is ambiguous 

437 scoring_weights: Weights for semantic/section/type signals 

438 exclude_common_files: Infrastructure files excluded from overlap detection 

439 """ 

440 

441 overlap_min_files: int = 2 

442 overlap_min_ratio: float = 0.25 

443 min_directory_depth: int = 2 

444 conflict_threshold: float = 0.4 

445 high_conflict_threshold: float = 0.7 

446 confidence_modifier: float = 0.5 

447 scoring_weights: ScoringWeightsConfig = field(default_factory=ScoringWeightsConfig) 

448 exclude_common_files: list[str] = field( 

449 default_factory=lambda: [ 

450 "__init__.py", 

451 "pyproject.toml", 

452 "setup.py", 

453 "setup.cfg", 

454 "CHANGELOG.md", 

455 "README.md", 

456 "conftest.py", 

457 ] 

458 ) 

459 

460 @classmethod 

461 def from_dict(cls, data: dict[str, Any]) -> DependencyMappingConfig: 

462 """Create DependencyMappingConfig from dictionary.""" 

463 return cls( 

464 overlap_min_files=data.get("overlap_min_files", 2), 

465 overlap_min_ratio=data.get("overlap_min_ratio", 0.25), 

466 min_directory_depth=data.get("min_directory_depth", 2), 

467 conflict_threshold=data.get("conflict_threshold", 0.4), 

468 high_conflict_threshold=data.get("high_conflict_threshold", 0.7), 

469 confidence_modifier=data.get("confidence_modifier", 0.5), 

470 scoring_weights=ScoringWeightsConfig.from_dict(data.get("scoring_weights", {})), 

471 exclude_common_files=data.get( 

472 "exclude_common_files", 

473 [ 

474 "__init__.py", 

475 "pyproject.toml", 

476 "setup.py", 

477 "setup.cfg", 

478 "CHANGELOG.md", 

479 "README.md", 

480 "conftest.py", 

481 ], 

482 ), 

483 ) 

484 

485 

486class BRConfig: 

487 """Main configuration class for little-loops. 

488 

489 Loads configuration from .claude/ll-config.json and merges with defaults. 

490 Provides convenient property access to all configuration values. 

491 

492 Example: 

493 config = BRConfig(Path.cwd()) 

494 print(config.project.src_dir) # "src/" 

495 print(config.issues.base_dir) # ".issues" 

496 print(config.get_issue_dir("bugs")) # Path(".issues/bugs") 

497 """ 

498 

499 CONFIG_FILENAME = "ll-config.json" 

500 CONFIG_DIR = ".claude" 

501 

502 def __init__(self, project_root: Path) -> None: 

503 """Initialize configuration from project root. 

504 

505 Args: 

506 project_root: Path to the project root directory 

507 """ 

508 self.project_root = project_root.resolve() 

509 self._raw_config = self._load_config() 

510 self._parse_config() 

511 

512 def _load_config(self) -> dict[str, Any]: 

513 """Load configuration from file.""" 

514 config_path = self.project_root / self.CONFIG_DIR / self.CONFIG_FILENAME 

515 if config_path.exists(): 

516 with open(config_path, encoding="utf-8") as f: 

517 return cast(dict[str, Any], json.load(f)) 

518 return {} 

519 

520 def _parse_config(self) -> None: 

521 """Parse raw config into typed dataclasses.""" 

522 self._project = ProjectConfig.from_dict(self._raw_config.get("project", {})) 

523 if not self._project.name: 

524 self._project.name = self.project_root.name 

525 

526 self._issues = IssuesConfig.from_dict(self._raw_config.get("issues", {})) 

527 self._automation = AutomationConfig.from_dict(self._raw_config.get("automation", {})) 

528 self._parallel = ParallelAutomationConfig.from_dict(self._raw_config.get("parallel", {})) 

529 self._commands = CommandsConfig.from_dict(self._raw_config.get("commands", {})) 

530 self._scan = ScanConfig.from_dict(self._raw_config.get("scan", {})) 

531 self._sprints = SprintsConfig.from_dict(self._raw_config.get("sprints", {})) 

532 self._loops = LoopsConfig.from_dict(self._raw_config.get("loops", {})) 

533 self._sync = SyncConfig.from_dict(self._raw_config.get("sync", {})) 

534 self._dependency_mapping = DependencyMappingConfig.from_dict( 

535 self._raw_config.get("dependency_mapping", {}) 

536 ) 

537 

538 @property 

539 def project(self) -> ProjectConfig: 

540 """Get project configuration.""" 

541 return self._project 

542 

543 @property 

544 def issues(self) -> IssuesConfig: 

545 """Get issues configuration.""" 

546 return self._issues 

547 

548 @property 

549 def automation(self) -> AutomationConfig: 

550 """Get automation configuration.""" 

551 return self._automation 

552 

553 @property 

554 def parallel(self) -> ParallelAutomationConfig: 

555 """Get parallel automation configuration.""" 

556 return self._parallel 

557 

558 @property 

559 def commands(self) -> CommandsConfig: 

560 """Get commands configuration.""" 

561 return self._commands 

562 

563 @property 

564 def scan(self) -> ScanConfig: 

565 """Get scan configuration.""" 

566 return self._scan 

567 

568 @property 

569 def sprints(self) -> SprintsConfig: 

570 """Get sprints configuration.""" 

571 return self._sprints 

572 

573 @property 

574 def loops(self) -> LoopsConfig: 

575 """Get loops configuration.""" 

576 return self._loops 

577 

578 @property 

579 def sync(self) -> SyncConfig: 

580 """Get sync configuration.""" 

581 return self._sync 

582 

583 @property 

584 def dependency_mapping(self) -> DependencyMappingConfig: 

585 """Get dependency mapping configuration.""" 

586 return self._dependency_mapping 

587 

588 @property 

589 def repo_path(self) -> Path: 

590 """Get the repository root path.""" 

591 return self.project_root 

592 

593 # Convenience methods for common operations 

594 

595 def get_issue_dir(self, category: str) -> Path: 

596 """Get the directory path for an issue category. 

597 

598 Args: 

599 category: Category key (e.g., "bugs", "features") 

600 

601 Returns: 

602 Path to the issue category directory 

603 """ 

604 if category in self._issues.categories: 

605 dir_name = self._issues.categories[category].dir 

606 else: 

607 dir_name = category 

608 return self.project_root / self._issues.base_dir / dir_name 

609 

610 def get_completed_dir(self) -> Path: 

611 """Get the path to the completed issues directory.""" 

612 return self.project_root / self._issues.base_dir / self._issues.completed_dir 

613 

614 def get_deferred_dir(self) -> Path: 

615 """Get the path to the deferred issues directory.""" 

616 return self.project_root / self._issues.base_dir / self._issues.deferred_dir 

617 

618 def get_issue_prefix(self, category: str) -> str: 

619 """Get the issue ID prefix for a category. 

620 

621 Args: 

622 category: Category key (e.g., "bugs", "features") 

623 

624 Returns: 

625 Issue prefix (e.g., "BUG", "FEAT") 

626 """ 

627 if category in self._issues.categories: 

628 return self._issues.categories[category].prefix 

629 return category.upper()[:3] 

630 

631 def get_category_action(self, category: str) -> str: 

632 """Get the default action for a category. 

633 

634 Args: 

635 category: Category key (e.g., "bugs", "features") 

636 

637 Returns: 

638 Action verb (e.g., "fix", "implement") 

639 """ 

640 if category in self._issues.categories: 

641 return self._issues.categories[category].action 

642 return "fix" 

643 

644 def get_loops_dir(self) -> Path: 

645 """Get the loops directory path.""" 

646 return self.project_root / self._loops.loops_dir 

647 

648 def get_src_path(self) -> Path: 

649 """Get the source directory path.""" 

650 return self.project_root / self._project.src_dir 

651 

652 def get_worktree_base(self) -> Path: 

653 """Get the worktree base directory path.""" 

654 return self.project_root / self._automation.worktree_base 

655 

656 def get_state_file(self) -> Path: 

657 """Get the state file path.""" 

658 return self.project_root / self._automation.state_file 

659 

660 def get_parallel_state_file(self) -> Path: 

661 """Get the parallel state file path.""" 

662 return self.project_root / self._parallel.base.state_file 

663 

664 def create_parallel_config( 

665 self, 

666 *, 

667 max_workers: int | None = None, 

668 priority_filter: list[str] | None = None, 

669 max_issues: int = 0, 

670 dry_run: bool = False, 

671 timeout_seconds: int | None = None, 

672 stream_output: bool | None = None, 

673 show_model: bool | None = None, 

674 only_ids: set[str] | None = None, 

675 skip_ids: set[str] | None = None, 

676 type_prefixes: set[str] | None = None, 

677 merge_pending: bool = False, 

678 clean_start: bool = False, 

679 ignore_pending: bool = False, 

680 overlap_detection: bool = False, 

681 serialize_overlapping: bool = True, 

682 base_branch: str = "main", 

683 ) -> ParallelConfig: 

684 """Create a ParallelConfig from BRConfig settings with optional overrides. 

685 

686 Args: 

687 max_workers: Override max_workers (default: from config) 

688 priority_filter: Override priority filter (default: from issues config) 

689 max_issues: Maximum issues to process (default: 0 = unlimited) 

690 dry_run: Preview mode (default: False) 

691 timeout_seconds: Per-issue timeout (default: from config) 

692 stream_output: Stream output (default: from config) 

693 show_model: Make API call to verify model (default: False) 

694 only_ids: If provided, only process these issue IDs 

695 skip_ids: Issue IDs to skip (in addition to completed/failed) 

696 merge_pending: Attempt to merge pending worktrees (default: False) 

697 clean_start: Remove all worktrees without checking (default: False) 

698 ignore_pending: Report pending work but continue (default: False) 

699 overlap_detection: Enable pre-flight overlap detection (default: False) 

700 serialize_overlapping: If True, defer overlapping issues; if False, just warn 

701 

702 Returns: 

703 ParallelConfig configured from BRConfig 

704 """ 

705 from little_loops.parallel.types import ParallelConfig 

706 

707 return ParallelConfig( 

708 max_workers=max_workers or self._parallel.base.max_workers, 

709 p0_sequential=self._parallel.p0_sequential, 

710 worktree_base=Path(self._parallel.base.worktree_base), 

711 state_file=Path(self._parallel.base.state_file), 

712 max_merge_retries=self._parallel.max_merge_retries, 

713 priority_filter=priority_filter or self._issues.priorities, 

714 max_issues=max_issues, 

715 dry_run=dry_run, 

716 timeout_per_issue=timeout_seconds or self._parallel.base.timeout_seconds, 

717 stream_subprocess_output=( 

718 stream_output if stream_output is not None else self._parallel.base.stream_output 

719 ), 

720 show_model=show_model if show_model is not None else False, 

721 command_prefix=self._parallel.command_prefix, 

722 ready_command=self._parallel.ready_command, 

723 manage_command=self._parallel.manage_command, 

724 only_ids=only_ids, 

725 skip_ids=skip_ids, 

726 type_prefixes=type_prefixes, 

727 worktree_copy_files=self._parallel.worktree_copy_files, 

728 require_code_changes=self._parallel.require_code_changes, 

729 merge_pending=merge_pending, 

730 clean_start=clean_start, 

731 ignore_pending=ignore_pending, 

732 overlap_detection=overlap_detection, 

733 serialize_overlapping=serialize_overlapping, 

734 base_branch=base_branch, 

735 ) 

736 

737 @property 

738 def issue_categories(self) -> list[str]: 

739 """Get list of configured issue category names.""" 

740 return list(self._issues.categories.keys()) 

741 

742 @property 

743 def issue_priorities(self) -> list[str]: 

744 """Get list of valid priority prefixes.""" 

745 return self._issues.priorities 

746 

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

748 """Convert configuration to dictionary. 

749 

750 Useful for variable substitution in command templates. 

751 """ 

752 return { 

753 "project": { 

754 "name": self._project.name, 

755 "src_dir": self._project.src_dir, 

756 "test_cmd": self._project.test_cmd, 

757 "lint_cmd": self._project.lint_cmd, 

758 "type_cmd": self._project.type_cmd, 

759 "format_cmd": self._project.format_cmd, 

760 "build_cmd": self._project.build_cmd, 

761 "run_cmd": self._project.run_cmd, 

762 }, 

763 "issues": { 

764 "base_dir": self._issues.base_dir, 

765 "categories": { 

766 k: {"prefix": v.prefix, "dir": v.dir, "action": v.action} 

767 for k, v in self._issues.categories.items() 

768 }, 

769 "completed_dir": self._issues.completed_dir, 

770 "deferred_dir": self._issues.deferred_dir, 

771 "priorities": self._issues.priorities, 

772 "templates_dir": self._issues.templates_dir, 

773 }, 

774 "automation": { 

775 "timeout_seconds": self._automation.timeout_seconds, 

776 "state_file": self._automation.state_file, 

777 "worktree_base": self._automation.worktree_base, 

778 "max_workers": self._automation.max_workers, 

779 "stream_output": self._automation.stream_output, 

780 "max_continuations": self._automation.max_continuations, 

781 }, 

782 "parallel": { 

783 "max_workers": self._parallel.base.max_workers, 

784 "p0_sequential": self._parallel.p0_sequential, 

785 "worktree_base": self._parallel.base.worktree_base, 

786 "state_file": self._parallel.base.state_file, 

787 "timeout_seconds": self._parallel.base.timeout_seconds, 

788 "max_merge_retries": self._parallel.max_merge_retries, 

789 "stream_output": self._parallel.base.stream_output, 

790 "command_prefix": self._parallel.command_prefix, 

791 "ready_command": self._parallel.ready_command, 

792 "manage_command": self._parallel.manage_command, 

793 }, 

794 "commands": { 

795 "pre_implement": self._commands.pre_implement, 

796 "post_implement": self._commands.post_implement, 

797 "custom_verification": self._commands.custom_verification, 

798 "confidence_gate": { 

799 "enabled": self._commands.confidence_gate.enabled, 

800 "threshold": self._commands.confidence_gate.threshold, 

801 }, 

802 "tdd_mode": self._commands.tdd_mode, 

803 }, 

804 "scan": { 

805 "focus_dirs": self._scan.focus_dirs, 

806 "exclude_patterns": self._scan.exclude_patterns, 

807 "custom_agents": self._scan.custom_agents, 

808 }, 

809 "sprints": { 

810 "sprints_dir": self._sprints.sprints_dir, 

811 "default_timeout": self._sprints.default_timeout, 

812 "default_max_workers": self._sprints.default_max_workers, 

813 }, 

814 "loops": { 

815 "loops_dir": self._loops.loops_dir, 

816 }, 

817 "sync": { 

818 "enabled": self._sync.enabled, 

819 "provider": self._sync.provider, 

820 "github": { 

821 "repo": self._sync.github.repo, 

822 "label_mapping": self._sync.github.label_mapping, 

823 "priority_labels": self._sync.github.priority_labels, 

824 "sync_completed": self._sync.github.sync_completed, 

825 "state_file": self._sync.github.state_file, 

826 "pull_template": self._sync.github.pull_template, 

827 }, 

828 }, 

829 "dependency_mapping": { 

830 "overlap_min_files": self._dependency_mapping.overlap_min_files, 

831 "overlap_min_ratio": self._dependency_mapping.overlap_min_ratio, 

832 "min_directory_depth": self._dependency_mapping.min_directory_depth, 

833 "conflict_threshold": self._dependency_mapping.conflict_threshold, 

834 "high_conflict_threshold": self._dependency_mapping.high_conflict_threshold, 

835 "confidence_modifier": self._dependency_mapping.confidence_modifier, 

836 "scoring_weights": { 

837 "semantic": self._dependency_mapping.scoring_weights.semantic, 

838 "section": self._dependency_mapping.scoring_weights.section, 

839 "type": self._dependency_mapping.scoring_weights.type, 

840 }, 

841 "exclude_common_files": self._dependency_mapping.exclude_common_files, 

842 }, 

843 } 

844 

845 def resolve_variable(self, var_path: str) -> str | None: 

846 """Resolve a variable path like 'project.src_dir' to its value. 

847 

848 Args: 

849 var_path: Dot-separated path to configuration value 

850 

851 Returns: 

852 The resolved value as a string, or None if not found 

853 """ 

854 parts = var_path.split(".") 

855 value: Any = self.to_dict() 

856 

857 for part in parts: 

858 if isinstance(value, dict) and part in value: 

859 value = value[part] 

860 else: 

861 return None 

862 

863 if value is None: 

864 return None 

865 if isinstance(value, list): 

866 return " ".join(str(v) for v in value) 

867 return str(value) 

868 

869 

870# Backwards compatibility alias 

871CLConfig = BRConfig