Coverage for little_loops / config / core.py: 51%
154 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -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
10import warnings
11from dataclasses import dataclass
12from pathlib import Path
13from typing import Any, cast
15from little_loops.config.automation import (
16 AutomationConfig,
17 CommandsConfig,
18 DependencyMappingConfig,
19 ParallelAutomationConfig,
20)
21from little_loops.config.cli import CliConfig, RefineStatusConfig
22from little_loops.config.features import (
23 EventsConfig,
24 IssuesConfig,
25 LearningTestsConfig,
26 LoopsConfig,
27 ScanConfig,
28 SprintsConfig,
29 SyncConfig,
30)
31from little_loops.parallel.types import ParallelConfig
34@dataclass
35class ProjectConfig:
36 """Project-level configuration."""
38 name: str = ""
39 src_dir: str = "src/"
40 test_dir: str = "tests"
41 test_cmd: str = "pytest"
42 lint_cmd: str = "ruff check ."
43 type_cmd: str | None = "mypy"
44 format_cmd: str | None = "ruff format ."
45 build_cmd: str | None = None
46 run_cmd: str | None = None
48 @classmethod
49 def from_dict(cls, data: dict[str, Any]) -> ProjectConfig:
50 """Create ProjectConfig from dictionary."""
51 return cls(
52 name=data.get("name", ""),
53 src_dir=data.get("src_dir", "src/"),
54 test_dir=data.get("test_dir", "tests"),
55 test_cmd=data.get("test_cmd", "pytest"),
56 lint_cmd=data.get("lint_cmd", "ruff check ."),
57 type_cmd=data.get("type_cmd", "mypy"),
58 format_cmd=data.get("format_cmd", "ruff format ."),
59 build_cmd=data.get("build_cmd"),
60 run_cmd=data.get("run_cmd"),
61 )
64class BRConfig:
65 """Main configuration class for little-loops.
67 Loads configuration from .ll/ll-config.json and merges with defaults.
68 Provides convenient property access to all configuration values.
70 Example:
71 config = BRConfig(Path.cwd())
72 print(config.project.src_dir) # "src/"
73 print(config.issues.base_dir) # ".issues"
74 print(config.get_issue_dir("bugs")) # Path(".issues/bugs")
75 """
77 CONFIG_FILENAME = "ll-config.json"
78 CONFIG_DIR = ".ll"
80 def __init__(self, project_root: Path) -> None:
81 """Initialize configuration from project root.
83 Args:
84 project_root: Path to the project root directory
85 """
86 self.project_root = project_root.resolve()
87 self._raw_config = self._load_config()
88 self._parse_config()
90 def _load_config(self) -> dict[str, Any]:
91 """Load configuration from file."""
92 config_path = self.project_root / self.CONFIG_DIR / self.CONFIG_FILENAME
93 if config_path.exists():
94 with open(config_path, encoding="utf-8") as f:
95 return cast(dict[str, Any], json.load(f))
96 return {}
98 def _parse_config(self) -> None:
99 """Parse raw config into typed dataclasses."""
100 self._project = ProjectConfig.from_dict(self._raw_config.get("project", {}))
101 if not self._project.name:
102 self._project.name = self.project_root.name
104 self._issues = IssuesConfig.from_dict(self._raw_config.get("issues", {}))
105 self._automation = AutomationConfig.from_dict(self._raw_config.get("automation", {}))
106 self._parallel = ParallelAutomationConfig.from_dict(self._raw_config.get("parallel", {}))
107 self._commands = CommandsConfig.from_dict(self._raw_config.get("commands", {}))
108 self._scan = ScanConfig.from_dict(self._raw_config.get("scan", {}))
109 self._sprints = SprintsConfig.from_dict(self._raw_config.get("sprints", {}))
110 self._loops = LoopsConfig.from_dict(self._raw_config.get("loops", {}))
111 self._learning_tests = LearningTestsConfig.from_dict(
112 self._raw_config.get("learning_tests", {})
113 )
114 self._sync = SyncConfig.from_dict(self._raw_config.get("sync", {}))
115 self._dependency_mapping = DependencyMappingConfig.from_dict(
116 self._raw_config.get("dependency_mapping", {})
117 )
118 self._cli = CliConfig.from_dict(self._raw_config.get("cli", {}))
119 self._refine_status = RefineStatusConfig.from_dict(
120 self._raw_config.get("refine_status", {})
121 )
122 self._events = EventsConfig.from_dict(self._raw_config.get("events", {}))
124 @property
125 def project(self) -> ProjectConfig:
126 """Get project configuration."""
127 return self._project
129 @property
130 def issues(self) -> IssuesConfig:
131 """Get issues configuration."""
132 return self._issues
134 @property
135 def automation(self) -> AutomationConfig:
136 """Get automation configuration."""
137 return self._automation
139 @property
140 def parallel(self) -> ParallelAutomationConfig:
141 """Get parallel automation configuration."""
142 return self._parallel
144 @property
145 def commands(self) -> CommandsConfig:
146 """Get commands configuration."""
147 return self._commands
149 @property
150 def scan(self) -> ScanConfig:
151 """Get scan configuration."""
152 return self._scan
154 @property
155 def sprints(self) -> SprintsConfig:
156 """Get sprints configuration."""
157 return self._sprints
159 @property
160 def loops(self) -> LoopsConfig:
161 """Get loops configuration."""
162 return self._loops
164 @property
165 def learning_tests(self) -> LearningTestsConfig:
166 """Get learning tests configuration."""
167 return self._learning_tests
169 @property
170 def sync(self) -> SyncConfig:
171 """Get sync configuration."""
172 return self._sync
174 @property
175 def dependency_mapping(self) -> DependencyMappingConfig:
176 """Get dependency mapping configuration."""
177 return self._dependency_mapping
179 @property
180 def cli(self) -> CliConfig:
181 """Get CLI output configuration."""
182 return self._cli
184 @property
185 def refine_status(self) -> RefineStatusConfig:
186 """Get refine-status display configuration."""
187 return self._refine_status
189 @property
190 def events(self) -> EventsConfig:
191 """Get events configuration."""
192 return self._events
194 @property
195 def extensions(self) -> list[str]:
196 """Get extension config paths (e.g. ``["module:Class", ...]``)."""
197 return self._raw_config.get("extensions", [])
199 @property
200 def repo_path(self) -> Path:
201 """Get the repository root path."""
202 return self.project_root
204 # Convenience methods for common operations
206 def get_issue_dir(self, category: str) -> Path:
207 """Get the directory path for an issue category.
209 Args:
210 category: Category key (e.g., "bugs", "features")
212 Returns:
213 Path to the issue category directory
214 """
215 if category in self._issues.categories:
216 dir_name = self._issues.categories[category].dir
217 else:
218 dir_name = category
219 return self.project_root / self._issues.base_dir / dir_name
221 def get_completed_dir(self) -> Path:
222 """Get the path to the completed issues directory."""
223 warnings.warn(
224 "BRConfig.get_completed_dir() is deprecated; use IssueInfo.status instead",
225 DeprecationWarning,
226 stacklevel=2,
227 )
228 return self.project_root / self._issues.base_dir / self._issues.completed_dir
230 def get_deferred_dir(self) -> Path:
231 """Get the path to the deferred issues directory."""
232 warnings.warn(
233 "BRConfig.get_deferred_dir() is deprecated; use IssueInfo.status instead",
234 DeprecationWarning,
235 stacklevel=2,
236 )
237 return self.project_root / self._issues.base_dir / self._issues.deferred_dir
239 def get_issue_prefix(self, category: str) -> str:
240 """Get the issue ID prefix for a category.
242 Args:
243 category: Category key (e.g., "bugs", "features")
245 Returns:
246 Issue prefix (e.g., "BUG", "FEAT")
247 """
248 if category in self._issues.categories:
249 return self._issues.categories[category].prefix
250 return category.upper()[:3]
252 def get_category_action(self, category: str) -> str:
253 """Get the default action for a category.
255 Args:
256 category: Category key (e.g., "bugs", "features")
258 Returns:
259 Action verb (e.g., "fix", "implement")
260 """
261 if category in self._issues.categories:
262 return self._issues.categories[category].action
263 return "fix"
265 def get_loops_dir(self) -> Path:
266 """Get the loops directory path."""
267 return self.project_root / self._loops.loops_dir
269 def get_src_path(self) -> Path:
270 """Get the source directory path."""
271 return self.project_root / self._project.src_dir
273 def get_worktree_base(self) -> Path:
274 """Get the worktree base directory path."""
275 return self.project_root / self._automation.worktree_base
277 def get_state_file(self) -> Path:
278 """Get the state file path."""
279 return self.project_root / self._automation.state_file
281 def get_parallel_state_file(self) -> Path:
282 """Get the parallel state file path."""
283 return self.project_root / self._parallel.base.state_file
285 def create_parallel_config(
286 self,
287 *,
288 max_workers: int | None = None,
289 priority_filter: list[str] | None = None,
290 max_issues: int = 0,
291 dry_run: bool = False,
292 timeout_seconds: int | None = None,
293 idle_timeout_per_issue: int | None = None,
294 stream_output: bool | None = None,
295 show_model: bool | None = None,
296 only_ids: set[str] | None = None,
297 skip_ids: set[str] | None = None,
298 type_prefixes: set[str] | None = None,
299 label_filter: set[str] | None = None,
300 merge_pending: bool = False,
301 clean_start: bool = False,
302 ignore_pending: bool = False,
303 overlap_detection: bool = False,
304 serialize_overlapping: bool = True,
305 base_branch: str = "main",
306 remote_name: str | None = None,
307 ) -> ParallelConfig:
308 """Create a ParallelConfig from BRConfig settings with optional overrides.
310 Args:
311 max_workers: Override max_workers (default: from config)
312 priority_filter: Override priority filter (default: from issues config)
313 max_issues: Maximum issues to process (default: 0 = unlimited)
314 dry_run: Preview mode (default: False)
315 timeout_seconds: Per-issue timeout (default: from config)
316 idle_timeout_per_issue: Kill worker if no output for N seconds (0 to disable, default: 0)
317 stream_output: Stream output (default: from config)
318 show_model: Make API call to verify model (default: False)
319 only_ids: If provided, only process these issue IDs
320 skip_ids: Issue IDs to skip (in addition to completed/failed)
321 merge_pending: Attempt to merge pending worktrees (default: False)
322 clean_start: Remove all worktrees without checking (default: False)
323 ignore_pending: Report pending work but continue (default: False)
324 overlap_detection: Enable pre-flight overlap detection (default: False)
325 serialize_overlapping: If True, defer overlapping issues; if False, just warn
327 Returns:
328 ParallelConfig configured from BRConfig
329 """
330 return ParallelConfig(
331 max_workers=max_workers or self._parallel.base.max_workers,
332 p0_sequential=self._parallel.p0_sequential,
333 worktree_base=Path(self._parallel.base.worktree_base),
334 state_file=Path(self._parallel.base.state_file),
335 max_merge_retries=self._parallel.max_merge_retries,
336 priority_filter=priority_filter or self._issues.priorities,
337 max_issues=max_issues,
338 dry_run=dry_run,
339 timeout_per_issue=timeout_seconds or self._parallel.base.timeout_seconds,
340 idle_timeout_per_issue=idle_timeout_per_issue
341 if idle_timeout_per_issue is not None
342 else 0,
343 stream_subprocess_output=(
344 stream_output if stream_output is not None else self._parallel.base.stream_output
345 ),
346 show_model=show_model if show_model is not None else False,
347 command_prefix=self._parallel.command_prefix,
348 ready_command=self._parallel.ready_command,
349 manage_command=self._parallel.manage_command,
350 decide_command=self._parallel.decide_command,
351 only_ids=only_ids,
352 skip_ids=skip_ids,
353 type_prefixes=type_prefixes,
354 label_filter=label_filter,
355 worktree_copy_files=self._parallel.worktree_copy_files,
356 require_code_changes=self._parallel.require_code_changes,
357 use_feature_branches=self._parallel.use_feature_branches,
358 merge_pending=merge_pending,
359 clean_start=clean_start,
360 ignore_pending=ignore_pending,
361 overlap_detection=overlap_detection,
362 serialize_overlapping=serialize_overlapping,
363 base_branch=base_branch,
364 remote_name=remote_name if remote_name is not None else self._parallel.remote_name,
365 )
367 @property
368 def issue_categories(self) -> list[str]:
369 """Get list of configured issue category names."""
370 return list(self._issues.categories.keys())
372 @property
373 def issue_priorities(self) -> list[str]:
374 """Get list of valid priority prefixes."""
375 return self._issues.priorities
377 def to_dict(self) -> dict[str, Any]:
378 """Convert configuration to dictionary.
380 Useful for variable substitution in command templates.
381 """
382 return {
383 "project": {
384 "name": self._project.name,
385 "src_dir": self._project.src_dir,
386 "test_dir": self._project.test_dir,
387 "test_cmd": self._project.test_cmd,
388 "lint_cmd": self._project.lint_cmd,
389 "type_cmd": self._project.type_cmd,
390 "format_cmd": self._project.format_cmd,
391 "build_cmd": self._project.build_cmd,
392 "run_cmd": self._project.run_cmd,
393 },
394 "issues": {
395 "base_dir": self._issues.base_dir,
396 "categories": {
397 k: {"prefix": v.prefix, "dir": v.dir, "action": v.action}
398 for k, v in self._issues.categories.items()
399 },
400 "priorities": self._issues.priorities,
401 "templates_dir": self._issues.templates_dir,
402 "capture_template": self._issues.capture_template,
403 },
404 "automation": {
405 "timeout_seconds": self._automation.timeout_seconds,
406 "idle_timeout_seconds": self._automation.idle_timeout_seconds,
407 "state_file": self._automation.state_file,
408 "worktree_base": self._automation.worktree_base,
409 "max_workers": self._automation.max_workers,
410 "stream_output": self._automation.stream_output,
411 "max_continuations": self._automation.max_continuations,
412 },
413 "parallel": {
414 "max_workers": self._parallel.base.max_workers,
415 "p0_sequential": self._parallel.p0_sequential,
416 "worktree_base": self._parallel.base.worktree_base,
417 "state_file": self._parallel.base.state_file,
418 "timeout_per_issue": self._parallel.base.timeout_seconds,
419 "max_merge_retries": self._parallel.max_merge_retries,
420 "stream_subprocess_output": self._parallel.base.stream_output,
421 "command_prefix": self._parallel.command_prefix,
422 "ready_command": self._parallel.ready_command,
423 "manage_command": self._parallel.manage_command,
424 "decide_command": self._parallel.decide_command,
425 "worktree_copy_files": self._parallel.worktree_copy_files,
426 "require_code_changes": self._parallel.require_code_changes,
427 "use_feature_branches": self._parallel.use_feature_branches,
428 "remote_name": self._parallel.remote_name,
429 },
430 "commands": {
431 "pre_implement": self._commands.pre_implement,
432 "post_implement": self._commands.post_implement,
433 "custom_verification": self._commands.custom_verification,
434 "confidence_gate": {
435 "enabled": self._commands.confidence_gate.enabled,
436 "readiness_threshold": self._commands.confidence_gate.readiness_threshold,
437 "outcome_threshold": self._commands.confidence_gate.outcome_threshold,
438 },
439 "tdd_mode": self._commands.tdd_mode,
440 "rate_limits": {
441 "max_wait_seconds": self._commands.rate_limits.max_wait_seconds,
442 "long_wait_ladder": self._commands.rate_limits.long_wait_ladder,
443 "circuit_breaker_enabled": self._commands.rate_limits.circuit_breaker_enabled,
444 "circuit_breaker_path": self._commands.rate_limits.circuit_breaker_path,
445 },
446 "recursive_refine": {
447 "max_depth": self._commands.recursive_refine.max_depth,
448 },
449 },
450 "scan": {
451 "focus_dirs": self._scan.focus_dirs,
452 "exclude_patterns": self._scan.exclude_patterns,
453 "custom_agents": self._scan.custom_agents,
454 },
455 "sprints": {
456 "sprints_dir": self._sprints.sprints_dir,
457 "default_timeout": self._sprints.default_timeout,
458 "default_max_workers": self._sprints.default_max_workers,
459 },
460 "loops": {
461 "loops_dir": self._loops.loops_dir,
462 "queue_wait_timeout_seconds": self._loops.queue_wait_timeout_seconds,
463 "glyphs": self._loops.glyphs.to_dict(),
464 },
465 "learning_tests": {
466 "stale_after_days": self._learning_tests.stale_after_days,
467 },
468 "events": {
469 "transports": list(self._events.transports),
470 "socket": {
471 "path": self._events.socket.path,
472 "max_clients": self._events.socket.max_clients,
473 },
474 "otel": {
475 "endpoint": self._events.otel.endpoint,
476 "service_name": self._events.otel.service_name,
477 },
478 "webhook": {
479 "url": self._events.webhook.url,
480 "batch_ms": self._events.webhook.batch_ms,
481 "headers": dict(self._events.webhook.headers),
482 },
483 },
484 "sync": {
485 "enabled": self._sync.enabled,
486 "provider": self._sync.provider,
487 "github": {
488 "repo": self._sync.github.repo,
489 "label_mapping": self._sync.github.label_mapping,
490 "priority_labels": self._sync.github.priority_labels,
491 "sync_completed": self._sync.github.sync_completed,
492 "state_file": self._sync.github.state_file,
493 "pull_template": self._sync.github.pull_template,
494 },
495 },
496 "dependency_mapping": {
497 "overlap_min_files": self._dependency_mapping.overlap_min_files,
498 "overlap_min_ratio": self._dependency_mapping.overlap_min_ratio,
499 "min_directory_depth": self._dependency_mapping.min_directory_depth,
500 "conflict_threshold": self._dependency_mapping.conflict_threshold,
501 "high_conflict_threshold": self._dependency_mapping.high_conflict_threshold,
502 "confidence_modifier": self._dependency_mapping.confidence_modifier,
503 "scoring_weights": {
504 "semantic": self._dependency_mapping.scoring_weights.semantic,
505 "section": self._dependency_mapping.scoring_weights.section,
506 "type": self._dependency_mapping.scoring_weights.type,
507 },
508 "exclude_common_files": self._dependency_mapping.exclude_common_files,
509 },
510 "cli": {
511 "color": self._cli.color,
512 "colors": {
513 "logger": {
514 "info": self._cli.colors.logger.info,
515 "success": self._cli.colors.logger.success,
516 "warning": self._cli.colors.logger.warning,
517 "error": self._cli.colors.logger.error,
518 },
519 "priority": {
520 "P0": self._cli.colors.priority.P0,
521 "P1": self._cli.colors.priority.P1,
522 "P2": self._cli.colors.priority.P2,
523 "P3": self._cli.colors.priority.P3,
524 "P4": self._cli.colors.priority.P4,
525 "P5": self._cli.colors.priority.P5,
526 },
527 "type": {
528 "BUG": self._cli.colors.type.BUG,
529 "FEAT": self._cli.colors.type.FEAT,
530 "ENH": self._cli.colors.type.ENH,
531 },
532 "fsm_active_state": self._cli.colors.fsm_active_state,
533 "fsm_edge_labels": {
534 "yes": self._cli.colors.fsm_edge_labels.yes,
535 "no": self._cli.colors.fsm_edge_labels.no,
536 "error": self._cli.colors.fsm_edge_labels.error,
537 "partial": self._cli.colors.fsm_edge_labels.partial,
538 "next": self._cli.colors.fsm_edge_labels.next,
539 "default": self._cli.colors.fsm_edge_labels.default,
540 "blocked": self._cli.colors.fsm_edge_labels.blocked,
541 "retry_exhausted": self._cli.colors.fsm_edge_labels.retry_exhausted,
542 "rate_limit_exhausted": self._cli.colors.fsm_edge_labels.rate_limit_exhausted,
543 },
544 },
545 },
546 }
548 def resolve_variable(self, var_path: str) -> str | None:
549 """Resolve a variable path like 'project.src_dir' to its value.
551 Args:
552 var_path: Dot-separated path to configuration value
554 Returns:
555 The resolved value as a string, or None if not found
556 """
557 parts = var_path.split(".")
558 value: Any = self.to_dict()
560 for part in parts:
561 if isinstance(value, dict) and part in value:
562 value = value[part]
563 else:
564 return None
566 if value is None:
567 return None
568 if isinstance(value, list):
569 return " ".join(str(v) for v in value)
570 return str(value)
573# Backwards compatibility alias
574CLConfig = BRConfig