Coverage for little_loops / config / automation.py: 100%
89 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-22 16:19 -0500
1"""Automation and execution configuration dataclasses.
3Covers automation scripts, parallel execution, confidence gates,
4command behavior, and dependency analysis configuration.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from typing import Any
13@dataclass
14class AutomationConfig:
15 """Automation script configuration."""
17 timeout_seconds: int = 3600
18 idle_timeout_seconds: int = 0 # Kill if no output for N seconds (0 to disable)
19 state_file: str = ".auto-manage-state.json"
20 worktree_base: str = ".worktrees"
21 max_workers: int = 2
22 stream_output: bool = True
23 max_continuations: int = 3 # Max session restarts on context handoff
25 @classmethod
26 def from_dict(cls, data: dict[str, Any]) -> AutomationConfig:
27 """Create AutomationConfig from dictionary."""
28 return cls(
29 timeout_seconds=data.get("timeout_seconds", 3600),
30 idle_timeout_seconds=data.get("idle_timeout_seconds", 0),
31 state_file=data.get("state_file", ".auto-manage-state.json"),
32 worktree_base=data.get("worktree_base", ".worktrees"),
33 max_workers=data.get("max_workers", 2),
34 stream_output=data.get("stream_output", True),
35 max_continuations=data.get("max_continuations", 3),
36 )
39@dataclass
40class ParallelAutomationConfig:
41 """Parallel automation configuration using composition.
43 Uses AutomationConfig for shared settings (max_workers, worktree_base,
44 state_file, timeout_seconds, stream_output) plus parallel-specific fields.
45 """
47 base: AutomationConfig
48 p0_sequential: bool = True
49 max_merge_retries: int = 2
50 command_prefix: str = "/ll:"
51 ready_command: str = "ready-issue {{issue_id}}"
52 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}"
53 decide_command: str = "decide-issue {{issue_id}}"
54 worktree_copy_files: list[str] = field(
55 default_factory=lambda: [".claude/settings.local.json", ".env"]
56 )
57 require_code_changes: bool = True
58 use_feature_branches: bool = False
59 remote_name: str = "origin"
61 @classmethod
62 def from_dict(cls, data: dict[str, Any]) -> ParallelAutomationConfig:
63 """Create ParallelAutomationConfig from dictionary.
65 Shared fields use parallel-specific defaults:
66 - state_file: ".parallel-manage-state.json"
67 - stream_output: False
68 """
69 base = AutomationConfig(
70 timeout_seconds=data.get("timeout_per_issue", data.get("timeout_seconds", 3600)),
71 state_file=data.get("state_file", ".parallel-manage-state.json"),
72 worktree_base=data.get("worktree_base", ".worktrees"),
73 max_workers=data.get("max_workers", 2),
74 stream_output=data.get("stream_subprocess_output", data.get("stream_output", False)),
75 )
76 return cls(
77 base=base,
78 p0_sequential=data.get("p0_sequential", True),
79 max_merge_retries=data.get("max_merge_retries", 2),
80 command_prefix=data.get("command_prefix", "/ll:"),
81 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"),
82 manage_command=data.get(
83 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}"
84 ),
85 decide_command=data.get("decide_command", "decide-issue {{issue_id}}"),
86 worktree_copy_files=data.get(
87 "worktree_copy_files", [".claude/settings.local.json", ".env"]
88 ),
89 require_code_changes=data.get("require_code_changes", True),
90 use_feature_branches=data.get("use_feature_branches", False),
91 remote_name=data.get("remote_name", "origin"),
92 )
95@dataclass
96class ConfidenceGateConfig:
97 """Confidence score gate configuration for manage-issue."""
99 enabled: bool = False
100 readiness_threshold: int = 85
101 outcome_threshold: int = 70
103 @classmethod
104 def from_dict(cls, data: dict[str, Any]) -> ConfidenceGateConfig:
105 """Create ConfidenceGateConfig from dictionary."""
106 legacy = data.get("threshold", 85)
107 return cls(
108 enabled=data.get("enabled", False),
109 readiness_threshold=data.get("readiness_threshold", legacy),
110 outcome_threshold=data.get("outcome_threshold", 70),
111 )
114@dataclass
115class RateLimitsConfig:
116 """Global rate-limit resilience configuration.
118 Defaults for per-state rate-limit handling when a state does not
119 override `rate_limit_max_wait_seconds` or `rate_limit_long_wait_ladder`,
120 plus shared circuit-breaker coordination settings used across parallel
121 worktrees.
123 Attributes:
124 max_wait_seconds: Total wall-clock budget for rate-limit handling
125 before routing to `on_rate_limit_exhausted`. Default 21600 (6h).
126 long_wait_ladder: Backoff ladder (seconds) for the long-wait tier
127 used once the short-tier retry budget is spent.
128 circuit_breaker_enabled: Whether the shared circuit breaker is active.
129 circuit_breaker_path: Path (relative to project root) for the shared
130 circuit-breaker state file.
131 """
133 max_wait_seconds: int = 21600
134 long_wait_ladder: list[int] = field(default_factory=lambda: [300, 900, 1800, 3600])
135 circuit_breaker_enabled: bool = True
136 circuit_breaker_path: str = ".loops/tmp/rate-limit-circuit.json"
138 @classmethod
139 def from_dict(cls, data: dict[str, Any]) -> RateLimitsConfig:
140 """Create RateLimitsConfig from dictionary."""
141 return cls(
142 max_wait_seconds=data.get("max_wait_seconds", 21600),
143 long_wait_ladder=data.get("long_wait_ladder", [300, 900, 1800, 3600]),
144 circuit_breaker_enabled=data.get("circuit_breaker_enabled", True),
145 circuit_breaker_path=data.get(
146 "circuit_breaker_path", ".loops/tmp/rate-limit-circuit.json"
147 ),
148 )
151@dataclass
152class RecursiveRefineConfig:
153 """Configuration for the recursive-refine loop."""
155 max_depth: int = 3
157 @classmethod
158 def from_dict(cls, data: dict[str, Any]) -> RecursiveRefineConfig:
159 """Create RecursiveRefineConfig from dictionary."""
160 return cls(max_depth=data.get("max_depth", 3))
163@dataclass
164class CommandsConfig:
165 """Command customization configuration."""
167 pre_implement: str | None = None
168 post_implement: str | None = None
169 custom_verification: list[str] = field(default_factory=list)
170 confidence_gate: ConfidenceGateConfig = field(default_factory=ConfidenceGateConfig)
171 tdd_mode: bool = False
172 rate_limits: RateLimitsConfig = field(default_factory=RateLimitsConfig)
173 recursive_refine: RecursiveRefineConfig = field(default_factory=RecursiveRefineConfig)
175 @classmethod
176 def from_dict(cls, data: dict[str, Any]) -> CommandsConfig:
177 """Create CommandsConfig from dictionary."""
178 return cls(
179 pre_implement=data.get("pre_implement"),
180 post_implement=data.get("post_implement"),
181 custom_verification=data.get("custom_verification", []),
182 confidence_gate=ConfidenceGateConfig.from_dict(data.get("confidence_gate", {})),
183 tdd_mode=data.get("tdd_mode", False),
184 rate_limits=RateLimitsConfig.from_dict(data.get("rate_limits", {})),
185 recursive_refine=RecursiveRefineConfig.from_dict(data.get("recursive_refine", {})),
186 )
189@dataclass
190class ScoringWeightsConfig:
191 """Scoring weights for semantic conflict analysis.
193 Weights for the three signals used in compute_conflict_score().
194 Should sum to 1.0 for normalized scoring.
196 Attributes:
197 semantic: Weight for semantic target overlap (component/function names)
198 section: Weight for section mention overlap (UI regions)
199 type: Weight for modification type match
200 """
202 semantic: float = 0.5
203 section: float = 0.3
204 type: float = 0.2
206 @classmethod
207 def from_dict(cls, data: dict[str, Any]) -> ScoringWeightsConfig:
208 """Create ScoringWeightsConfig from dictionary."""
209 return cls(
210 semantic=data.get("semantic", 0.5),
211 section=data.get("section", 0.3),
212 type=data.get("type", 0.2),
213 )
216@dataclass
217class DependencyMappingConfig:
218 """Dependency mapping threshold configuration.
220 Controls overlap detection sensitivity and conflict scoring thresholds.
221 Default values match the previously hardcoded constants for backwards
222 compatibility.
224 Attributes:
225 overlap_min_files: Minimum overlapping files to trigger overlap
226 overlap_min_ratio: Minimum ratio of overlapping files to smaller set
227 min_directory_depth: Minimum path segments for directory overlap
228 conflict_threshold: Below = parallel-safe, above = dependency proposed
229 high_conflict_threshold: Above = HIGH conflict label
230 confidence_modifier: Applied when dependency direction is ambiguous
231 scoring_weights: Weights for semantic/section/type signals
232 exclude_common_files: Infrastructure files excluded from overlap detection
233 """
235 overlap_min_files: int = 2
236 overlap_min_ratio: float = 0.25
237 min_directory_depth: int = 2
238 conflict_threshold: float = 0.4
239 high_conflict_threshold: float = 0.7
240 confidence_modifier: float = 0.5
241 scoring_weights: ScoringWeightsConfig = field(default_factory=ScoringWeightsConfig)
242 exclude_common_files: list[str] = field(
243 default_factory=lambda: [
244 "__init__.py",
245 "pyproject.toml",
246 "setup.py",
247 "setup.cfg",
248 "CHANGELOG.md",
249 "README.md",
250 "conftest.py",
251 ]
252 )
254 @classmethod
255 def from_dict(cls, data: dict[str, Any]) -> DependencyMappingConfig:
256 """Create DependencyMappingConfig from dictionary."""
257 return cls(
258 overlap_min_files=data.get("overlap_min_files", 2),
259 overlap_min_ratio=data.get("overlap_min_ratio", 0.25),
260 min_directory_depth=data.get("min_directory_depth", 2),
261 conflict_threshold=data.get("conflict_threshold", 0.4),
262 high_conflict_threshold=data.get("high_conflict_threshold", 0.7),
263 confidence_modifier=data.get("confidence_modifier", 0.5),
264 scoring_weights=ScoringWeightsConfig.from_dict(data.get("scoring_weights", {})),
265 exclude_common_files=data.get(
266 "exclude_common_files",
267 [
268 "__init__.py",
269 "pyproject.toml",
270 "setup.py",
271 "setup.cfg",
272 "CHANGELOG.md",
273 "README.md",
274 "conftest.py",
275 ],
276 ),
277 )