Coverage for little_loops / config / core.py: 75%
143 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Core configuration dataclasses and the root BRConfig aggregator.
3ProjectConfig holds project-level settings. BRConfig is the single entry
4point that loads ll-config.json and exposes all domain configs via properties.
5"""
7from __future__ import annotations
9import json
10from dataclasses import dataclass
11from pathlib import Path
12from typing import Any, cast
14from little_loops.config.automation import (
15 AutomationConfig,
16 CommandsConfig,
17 DependencyMappingConfig,
18 ParallelAutomationConfig,
19)
20from little_loops.config.cli import CliConfig, RefineStatusConfig
21from little_loops.config.features import (
22 IssuesConfig,
23 LoopsConfig,
24 ScanConfig,
25 SprintsConfig,
26 SyncConfig,
27)
28from little_loops.parallel.types import ParallelConfig
31@dataclass
32class ProjectConfig:
33 """Project-level configuration."""
35 name: str = ""
36 src_dir: str = "src/"
37 test_dir: str = "tests"
38 test_cmd: str = "pytest"
39 lint_cmd: str = "ruff check ."
40 type_cmd: str | None = "mypy"
41 format_cmd: str | None = "ruff format ."
42 build_cmd: str | None = None
43 run_cmd: str | None = None
45 @classmethod
46 def from_dict(cls, data: dict[str, Any]) -> ProjectConfig:
47 """Create ProjectConfig from dictionary."""
48 return cls(
49 name=data.get("name", ""),
50 src_dir=data.get("src_dir", "src/"),
51 test_dir=data.get("test_dir", "tests"),
52 test_cmd=data.get("test_cmd", "pytest"),
53 lint_cmd=data.get("lint_cmd", "ruff check ."),
54 type_cmd=data.get("type_cmd", "mypy"),
55 format_cmd=data.get("format_cmd", "ruff format ."),
56 build_cmd=data.get("build_cmd"),
57 run_cmd=data.get("run_cmd"),
58 )
61class BRConfig:
62 """Main configuration class for little-loops.
64 Loads configuration from .ll/ll-config.json and merges with defaults.
65 Provides convenient property access to all configuration values.
67 Example:
68 config = BRConfig(Path.cwd())
69 print(config.project.src_dir) # "src/"
70 print(config.issues.base_dir) # ".issues"
71 print(config.get_issue_dir("bugs")) # Path(".issues/bugs")
72 """
74 CONFIG_FILENAME = "ll-config.json"
75 CONFIG_DIR = ".ll"
77 def __init__(self, project_root: Path) -> None:
78 """Initialize configuration from project root.
80 Args:
81 project_root: Path to the project root directory
82 """
83 self.project_root = project_root.resolve()
84 self._raw_config = self._load_config()
85 self._parse_config()
87 def _load_config(self) -> dict[str, Any]:
88 """Load configuration from file."""
89 config_path = self.project_root / self.CONFIG_DIR / self.CONFIG_FILENAME
90 if config_path.exists():
91 with open(config_path, encoding="utf-8") as f:
92 return cast(dict[str, Any], json.load(f))
93 return {}
95 def _parse_config(self) -> None:
96 """Parse raw config into typed dataclasses."""
97 self._project = ProjectConfig.from_dict(self._raw_config.get("project", {}))
98 if not self._project.name:
99 self._project.name = self.project_root.name
101 self._issues = IssuesConfig.from_dict(self._raw_config.get("issues", {}))
102 self._automation = AutomationConfig.from_dict(self._raw_config.get("automation", {}))
103 self._parallel = ParallelAutomationConfig.from_dict(self._raw_config.get("parallel", {}))
104 self._commands = CommandsConfig.from_dict(self._raw_config.get("commands", {}))
105 self._scan = ScanConfig.from_dict(self._raw_config.get("scan", {}))
106 self._sprints = SprintsConfig.from_dict(self._raw_config.get("sprints", {}))
107 self._loops = LoopsConfig.from_dict(self._raw_config.get("loops", {}))
108 self._sync = SyncConfig.from_dict(self._raw_config.get("sync", {}))
109 self._dependency_mapping = DependencyMappingConfig.from_dict(
110 self._raw_config.get("dependency_mapping", {})
111 )
112 self._cli = CliConfig.from_dict(self._raw_config.get("cli", {}))
113 self._refine_status = RefineStatusConfig.from_dict(
114 self._raw_config.get("refine_status", {})
115 )
117 @property
118 def project(self) -> ProjectConfig:
119 """Get project configuration."""
120 return self._project
122 @property
123 def issues(self) -> IssuesConfig:
124 """Get issues configuration."""
125 return self._issues
127 @property
128 def automation(self) -> AutomationConfig:
129 """Get automation configuration."""
130 return self._automation
132 @property
133 def parallel(self) -> ParallelAutomationConfig:
134 """Get parallel automation configuration."""
135 return self._parallel
137 @property
138 def commands(self) -> CommandsConfig:
139 """Get commands configuration."""
140 return self._commands
142 @property
143 def scan(self) -> ScanConfig:
144 """Get scan configuration."""
145 return self._scan
147 @property
148 def sprints(self) -> SprintsConfig:
149 """Get sprints configuration."""
150 return self._sprints
152 @property
153 def loops(self) -> LoopsConfig:
154 """Get loops configuration."""
155 return self._loops
157 @property
158 def sync(self) -> SyncConfig:
159 """Get sync configuration."""
160 return self._sync
162 @property
163 def dependency_mapping(self) -> DependencyMappingConfig:
164 """Get dependency mapping configuration."""
165 return self._dependency_mapping
167 @property
168 def cli(self) -> CliConfig:
169 """Get CLI output configuration."""
170 return self._cli
172 @property
173 def refine_status(self) -> RefineStatusConfig:
174 """Get refine-status display configuration."""
175 return self._refine_status
177 @property
178 def extensions(self) -> list[str]:
179 """Get extension config paths (e.g. ``["module:Class", ...]``)."""
180 return self._raw_config.get("extensions", [])
182 @property
183 def repo_path(self) -> Path:
184 """Get the repository root path."""
185 return self.project_root
187 # Convenience methods for common operations
189 def get_issue_dir(self, category: str) -> Path:
190 """Get the directory path for an issue category.
192 Args:
193 category: Category key (e.g., "bugs", "features")
195 Returns:
196 Path to the issue category directory
197 """
198 if category in self._issues.categories:
199 dir_name = self._issues.categories[category].dir
200 else:
201 dir_name = category
202 return self.project_root / self._issues.base_dir / dir_name
204 def get_completed_dir(self) -> Path:
205 """Get the path to the completed issues directory."""
206 return self.project_root / self._issues.base_dir / self._issues.completed_dir
208 def get_deferred_dir(self) -> Path:
209 """Get the path to the deferred issues directory."""
210 return self.project_root / self._issues.base_dir / self._issues.deferred_dir
212 def get_issue_prefix(self, category: str) -> str:
213 """Get the issue ID prefix for a category.
215 Args:
216 category: Category key (e.g., "bugs", "features")
218 Returns:
219 Issue prefix (e.g., "BUG", "FEAT")
220 """
221 if category in self._issues.categories:
222 return self._issues.categories[category].prefix
223 return category.upper()[:3]
225 def get_category_action(self, category: str) -> str:
226 """Get the default action for a category.
228 Args:
229 category: Category key (e.g., "bugs", "features")
231 Returns:
232 Action verb (e.g., "fix", "implement")
233 """
234 if category in self._issues.categories:
235 return self._issues.categories[category].action
236 return "fix"
238 def get_loops_dir(self) -> Path:
239 """Get the loops directory path."""
240 return self.project_root / self._loops.loops_dir
242 def get_src_path(self) -> Path:
243 """Get the source directory path."""
244 return self.project_root / self._project.src_dir
246 def get_worktree_base(self) -> Path:
247 """Get the worktree base directory path."""
248 return self.project_root / self._automation.worktree_base
250 def get_state_file(self) -> Path:
251 """Get the state file path."""
252 return self.project_root / self._automation.state_file
254 def get_parallel_state_file(self) -> Path:
255 """Get the parallel state file path."""
256 return self.project_root / self._parallel.base.state_file
258 def create_parallel_config(
259 self,
260 *,
261 max_workers: int | None = None,
262 priority_filter: list[str] | None = None,
263 max_issues: int = 0,
264 dry_run: bool = False,
265 timeout_seconds: int | None = None,
266 idle_timeout_per_issue: int | None = None,
267 stream_output: bool | None = None,
268 show_model: bool | None = None,
269 only_ids: set[str] | None = None,
270 skip_ids: set[str] | None = None,
271 type_prefixes: set[str] | None = None,
272 merge_pending: bool = False,
273 clean_start: bool = False,
274 ignore_pending: bool = False,
275 overlap_detection: bool = False,
276 serialize_overlapping: bool = True,
277 base_branch: str = "main",
278 remote_name: str | None = None,
279 ) -> ParallelConfig:
280 """Create a ParallelConfig from BRConfig settings with optional overrides.
282 Args:
283 max_workers: Override max_workers (default: from config)
284 priority_filter: Override priority filter (default: from issues config)
285 max_issues: Maximum issues to process (default: 0 = unlimited)
286 dry_run: Preview mode (default: False)
287 timeout_seconds: Per-issue timeout (default: from config)
288 idle_timeout_per_issue: Kill worker if no output for N seconds (0 to disable, default: 0)
289 stream_output: Stream output (default: from config)
290 show_model: Make API call to verify model (default: False)
291 only_ids: If provided, only process these issue IDs
292 skip_ids: Issue IDs to skip (in addition to completed/failed)
293 merge_pending: Attempt to merge pending worktrees (default: False)
294 clean_start: Remove all worktrees without checking (default: False)
295 ignore_pending: Report pending work but continue (default: False)
296 overlap_detection: Enable pre-flight overlap detection (default: False)
297 serialize_overlapping: If True, defer overlapping issues; if False, just warn
299 Returns:
300 ParallelConfig configured from BRConfig
301 """
302 return ParallelConfig(
303 max_workers=max_workers or self._parallel.base.max_workers,
304 p0_sequential=self._parallel.p0_sequential,
305 worktree_base=Path(self._parallel.base.worktree_base),
306 state_file=Path(self._parallel.base.state_file),
307 max_merge_retries=self._parallel.max_merge_retries,
308 priority_filter=priority_filter or self._issues.priorities,
309 max_issues=max_issues,
310 dry_run=dry_run,
311 timeout_per_issue=timeout_seconds or self._parallel.base.timeout_seconds,
312 idle_timeout_per_issue=idle_timeout_per_issue
313 if idle_timeout_per_issue is not None
314 else 0,
315 stream_subprocess_output=(
316 stream_output if stream_output is not None else self._parallel.base.stream_output
317 ),
318 show_model=show_model if show_model is not None else False,
319 command_prefix=self._parallel.command_prefix,
320 ready_command=self._parallel.ready_command,
321 manage_command=self._parallel.manage_command,
322 only_ids=only_ids,
323 skip_ids=skip_ids,
324 type_prefixes=type_prefixes,
325 worktree_copy_files=self._parallel.worktree_copy_files,
326 require_code_changes=self._parallel.require_code_changes,
327 use_feature_branches=self._parallel.use_feature_branches,
328 merge_pending=merge_pending,
329 clean_start=clean_start,
330 ignore_pending=ignore_pending,
331 overlap_detection=overlap_detection,
332 serialize_overlapping=serialize_overlapping,
333 base_branch=base_branch,
334 remote_name=remote_name if remote_name is not None else self._parallel.remote_name,
335 )
337 @property
338 def issue_categories(self) -> list[str]:
339 """Get list of configured issue category names."""
340 return list(self._issues.categories.keys())
342 @property
343 def issue_priorities(self) -> list[str]:
344 """Get list of valid priority prefixes."""
345 return self._issues.priorities
347 def to_dict(self) -> dict[str, Any]:
348 """Convert configuration to dictionary.
350 Useful for variable substitution in command templates.
351 """
352 return {
353 "project": {
354 "name": self._project.name,
355 "src_dir": self._project.src_dir,
356 "test_dir": self._project.test_dir,
357 "test_cmd": self._project.test_cmd,
358 "lint_cmd": self._project.lint_cmd,
359 "type_cmd": self._project.type_cmd,
360 "format_cmd": self._project.format_cmd,
361 "build_cmd": self._project.build_cmd,
362 "run_cmd": self._project.run_cmd,
363 },
364 "issues": {
365 "base_dir": self._issues.base_dir,
366 "categories": {
367 k: {"prefix": v.prefix, "dir": v.dir, "action": v.action}
368 for k, v in self._issues.categories.items()
369 },
370 "completed_dir": self._issues.completed_dir,
371 "deferred_dir": self._issues.deferred_dir,
372 "priorities": self._issues.priorities,
373 "templates_dir": self._issues.templates_dir,
374 "capture_template": self._issues.capture_template,
375 },
376 "automation": {
377 "timeout_seconds": self._automation.timeout_seconds,
378 "idle_timeout_seconds": self._automation.idle_timeout_seconds,
379 "state_file": self._automation.state_file,
380 "worktree_base": self._automation.worktree_base,
381 "max_workers": self._automation.max_workers,
382 "stream_output": self._automation.stream_output,
383 "max_continuations": self._automation.max_continuations,
384 },
385 "parallel": {
386 "max_workers": self._parallel.base.max_workers,
387 "p0_sequential": self._parallel.p0_sequential,
388 "worktree_base": self._parallel.base.worktree_base,
389 "state_file": self._parallel.base.state_file,
390 "timeout_per_issue": self._parallel.base.timeout_seconds,
391 "max_merge_retries": self._parallel.max_merge_retries,
392 "stream_subprocess_output": self._parallel.base.stream_output,
393 "command_prefix": self._parallel.command_prefix,
394 "ready_command": self._parallel.ready_command,
395 "manage_command": self._parallel.manage_command,
396 "worktree_copy_files": self._parallel.worktree_copy_files,
397 "require_code_changes": self._parallel.require_code_changes,
398 "use_feature_branches": self._parallel.use_feature_branches,
399 "remote_name": self._parallel.remote_name,
400 },
401 "commands": {
402 "pre_implement": self._commands.pre_implement,
403 "post_implement": self._commands.post_implement,
404 "custom_verification": self._commands.custom_verification,
405 "confidence_gate": {
406 "enabled": self._commands.confidence_gate.enabled,
407 "readiness_threshold": self._commands.confidence_gate.readiness_threshold,
408 "outcome_threshold": self._commands.confidence_gate.outcome_threshold,
409 },
410 "tdd_mode": self._commands.tdd_mode,
411 },
412 "scan": {
413 "focus_dirs": self._scan.focus_dirs,
414 "exclude_patterns": self._scan.exclude_patterns,
415 "custom_agents": self._scan.custom_agents,
416 },
417 "sprints": {
418 "sprints_dir": self._sprints.sprints_dir,
419 "default_timeout": self._sprints.default_timeout,
420 "default_max_workers": self._sprints.default_max_workers,
421 },
422 "loops": {
423 "loops_dir": self._loops.loops_dir,
424 "glyphs": self._loops.glyphs.to_dict(),
425 },
426 "sync": {
427 "enabled": self._sync.enabled,
428 "provider": self._sync.provider,
429 "github": {
430 "repo": self._sync.github.repo,
431 "label_mapping": self._sync.github.label_mapping,
432 "priority_labels": self._sync.github.priority_labels,
433 "sync_completed": self._sync.github.sync_completed,
434 "state_file": self._sync.github.state_file,
435 "pull_template": self._sync.github.pull_template,
436 },
437 },
438 "dependency_mapping": {
439 "overlap_min_files": self._dependency_mapping.overlap_min_files,
440 "overlap_min_ratio": self._dependency_mapping.overlap_min_ratio,
441 "min_directory_depth": self._dependency_mapping.min_directory_depth,
442 "conflict_threshold": self._dependency_mapping.conflict_threshold,
443 "high_conflict_threshold": self._dependency_mapping.high_conflict_threshold,
444 "confidence_modifier": self._dependency_mapping.confidence_modifier,
445 "scoring_weights": {
446 "semantic": self._dependency_mapping.scoring_weights.semantic,
447 "section": self._dependency_mapping.scoring_weights.section,
448 "type": self._dependency_mapping.scoring_weights.type,
449 },
450 "exclude_common_files": self._dependency_mapping.exclude_common_files,
451 },
452 "cli": {
453 "color": self._cli.color,
454 "colors": {
455 "logger": {
456 "info": self._cli.colors.logger.info,
457 "success": self._cli.colors.logger.success,
458 "warning": self._cli.colors.logger.warning,
459 "error": self._cli.colors.logger.error,
460 },
461 "priority": {
462 "P0": self._cli.colors.priority.P0,
463 "P1": self._cli.colors.priority.P1,
464 "P2": self._cli.colors.priority.P2,
465 "P3": self._cli.colors.priority.P3,
466 "P4": self._cli.colors.priority.P4,
467 "P5": self._cli.colors.priority.P5,
468 },
469 "type": {
470 "BUG": self._cli.colors.type.BUG,
471 "FEAT": self._cli.colors.type.FEAT,
472 "ENH": self._cli.colors.type.ENH,
473 },
474 "fsm_active_state": self._cli.colors.fsm_active_state,
475 "fsm_edge_labels": {
476 "yes": self._cli.colors.fsm_edge_labels.yes,
477 "no": self._cli.colors.fsm_edge_labels.no,
478 "error": self._cli.colors.fsm_edge_labels.error,
479 "partial": self._cli.colors.fsm_edge_labels.partial,
480 "next": self._cli.colors.fsm_edge_labels.next,
481 "default": self._cli.colors.fsm_edge_labels.default,
482 "blocked": self._cli.colors.fsm_edge_labels.blocked,
483 "retry_exhausted": self._cli.colors.fsm_edge_labels.retry_exhausted,
484 },
485 },
486 },
487 }
489 def resolve_variable(self, var_path: str) -> str | None:
490 """Resolve a variable path like 'project.src_dir' to its value.
492 Args:
493 var_path: Dot-separated path to configuration value
495 Returns:
496 The resolved value as a string, or None if not found
497 """
498 parts = var_path.split(".")
499 value: Any = self.to_dict()
501 for part in parts:
502 if isinstance(value, dict) and part in value:
503 value = value[part]
504 else:
505 return None
507 if value is None:
508 return None
509 if isinstance(value, list):
510 return " ".join(str(v) for v in value)
511 return str(value)
514# Backwards compatibility alias
515CLConfig = BRConfig