Coverage for little_loops / sprint.py: 0%

1"""Sprint and sequence management for issue execution.""" 

2 

3import logging 

4import re 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from pathlib import Path 

8from typing import TYPE_CHECKING, Any 

9 

10import yaml 

11 

12logger = logging.getLogger(__name__) 

13 

14_EPIC_ID_RE = re.compile(r"^EPIC-\d+$", re.IGNORECASE) 

15_ACTIVE_STATUSES: set[str] = {"open", "in_progress", "blocked"} 

16 

17if TYPE_CHECKING: 

18 from little_loops.config import BRConfig 

19 from little_loops.issue_parser import IssueInfo 

20 

21 

22@dataclass 

23class SprintOptions: 

24 """Execution options for sprint runs. 

25 

26 Attributes: 

27 max_iterations: Maximum Claude iterations per issue 

28 timeout: Per-issue timeout in seconds 

29 max_workers: Worker count for parallel execution within waves 

30 """ 

31 

32 max_iterations: int = 100 

33 timeout: int = 3600 

34 max_workers: int = 2 

35 

36 def to_dict(self) -> dict: 

37 """Convert to dictionary for YAML serialization.""" 

38 return { 

39 "max_iterations": self.max_iterations, 

40 "timeout": self.timeout, 

41 "max_workers": self.max_workers, 

42 } 

43 

44 @classmethod 

45 def from_dict(cls, data: dict | None) -> "SprintOptions": 

46 """Create from dictionary (YAML deserialization). 

47 

48 Args: 

49 data: Dictionary from YAML file or None for defaults 

50 

51 Returns: 

52 SprintOptions instance 

53 """ 

54 if data is None: 

55 return cls() 

56 return cls( 

57 max_iterations=data.get("max_iterations", 100), 

58 timeout=data.get("timeout", 3600), 

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

60 ) 

61 

62 

63@dataclass 

64class SprintState: 

65 """Persistent state for sprint execution. 

66 

67 Enables resume capability after interruption by tracking: 

68 - Sprint name being executed 

69 - Current wave number 

70 - Completed issues 

71 - Failed issues with reasons 

72 - Timing information 

73 

74 Attributes: 

75 sprint_name: Name of the sprint being executed 

76 current_wave: Wave number currently being processed (1-indexed) 

77 completed_issues: List of completed issue IDs 

78 failed_issues: Mapping of issue ID to failure reason 

79 timing: Per-issue timing breakdown 

80 started_at: ISO 8601 timestamp when sprint started 

81 last_checkpoint: ISO 8601 timestamp of last state save 

82 """ 

83 

84 sprint_name: str = "" 

85 current_wave: int = 0 

86 completed_issues: list[str] = field(default_factory=list) 

87 failed_issues: dict[str, str] = field(default_factory=dict) 

88 skipped_blocked_issues: dict[str, str] = field(default_factory=dict) 

89 timing: dict[str, dict[str, float]] = field(default_factory=dict) 

90 started_at: str = "" 

91 last_checkpoint: str = "" 

92 

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

94 """Convert state to dictionary for JSON serialization.""" 

95 return { 

96 "sprint_name": self.sprint_name, 

97 "current_wave": self.current_wave, 

98 "completed_issues": self.completed_issues, 

99 "failed_issues": self.failed_issues, 

100 "skipped_blocked_issues": self.skipped_blocked_issues, 

101 "timing": self.timing, 

102 "started_at": self.started_at, 

103 "last_checkpoint": self.last_checkpoint, 

104 } 

105 

106 @classmethod 

107 def from_dict(cls, data: dict[str, Any]) -> "SprintState": 

108 """Create state from dictionary (JSON deserialization).""" 

109 return cls( 

110 sprint_name=data.get("sprint_name", ""), 

111 current_wave=data.get("current_wave", 0), 

112 completed_issues=data.get("completed_issues", []), 

113 failed_issues=data.get("failed_issues", {}), 

114 skipped_blocked_issues=data.get("skipped_blocked_issues", {}), 

115 timing=data.get("timing", {}), 

116 started_at=data.get("started_at", ""), 

117 last_checkpoint=data.get("last_checkpoint", ""), 

118 ) 

119 

120 

121@dataclass 

122class Sprint: 

123 """A sprint is a named group of issues to execute together. 

124 

125 Sprints allow planning work in batches and executing them as a unit. 

126 Execution is always dependency-aware with parallel waves. 

127 

128 Attributes: 

129 name: Sprint identifier (used as filename) 

130 description: Human-readable purpose 

131 issues: List of issue IDs (e.g., BUG-001, FEAT-010) 

132 created: ISO 8601 timestamp of creation 

133 options: Execution options (timeout, max_workers, etc.) 

134 """ 

135 

136 name: str 

137 description: str 

138 issues: list[str] 

139 created: str 

140 options: SprintOptions | None = None 

141 

142 def to_dict(self) -> dict[str, str | list[str] | dict]: 

143 """Convert to dictionary for YAML serialization. 

144 

145 Returns: 

146 Dictionary representation suitable for yaml.dump() 

147 """ 

148 data: dict[str, str | list[str] | dict] = { 

149 "name": self.name, 

150 "description": self.description, 

151 "created": self.created, 

152 "issues": self.issues, 

153 } 

154 if self.options: 

155 data["options"] = self.options.to_dict() 

156 return data 

157 

158 @classmethod 

159 def from_dict(cls, data: dict) -> "Sprint": 

160 """Create from dictionary (YAML deserialization). 

161 

162 Args: 

163 data: Dictionary from YAML file 

164 

165 Returns: 

166 Sprint instance 

167 """ 

168 return cls( 

169 name=data["name"], 

170 description=data.get("description", ""), 

171 issues=data.get("issues", []), 

172 created=data.get("created", datetime.now(UTC).isoformat()), 

173 options=SprintOptions.from_dict(data.get("options")), 

174 ) 

175 

176 def save(self, sprints_dir: Path) -> Path: 

177 """Save sprint to YAML file. 

178 

179 Args: 

180 sprints_dir: Directory containing sprint definitions 

181 

182 Returns: 

183 Path to saved file 

184 """ 

185 sprints_dir.mkdir(parents=True, exist_ok=True) 

186 sprint_path = sprints_dir / f"{self.name}.yaml" 

187 with open(sprint_path, "w") as f: 

188 yaml.dump(self.to_dict(), f, default_flow_style=False, sort_keys=False) 

189 return sprint_path 

190 

191 @classmethod 

192 def load(cls, sprints_dir: Path, name: str) -> "Sprint | None": 

193 """Load sprint from YAML file. 

194 

195 Args: 

196 sprints_dir: Directory containing sprint definitions 

197 name: Sprint name (without .yaml extension) 

198 

199 Returns: 

200 Sprint instance or None if not found 

201 """ 

202 sprint_path = sprints_dir / f"{name}.yaml" 

203 if not sprint_path.exists(): 

204 return None 

205 with open(sprint_path) as f: 

206 data = yaml.safe_load(f) 

207 return cls.from_dict(data) 

208 

209 

210class SprintManager: 

211 """Manager for sprint CRUD operations. 

212 

213 Provides methods to create, load, list, and delete sprint definitions. 

214 Also validates that issue IDs exist before executing sprints. 

215 """ 

216 

217 def __init__(self, sprints_dir: Path | None = None, config: "BRConfig | None" = None) -> None: 

218 """Initialize SprintManager. 

219 

220 Args: 

221 sprints_dir: Directory for sprint definitions (overrides config) 

222 config: Project configuration for settings and issue validation 

223 """ 

224 self.config = config 

225 # Derive sprints_dir: explicit arg > config > default 

226 if sprints_dir is not None: 

227 self.sprints_dir = sprints_dir 

228 elif config is not None: 

229 self.sprints_dir = Path(config.sprints.sprints_dir) 

230 else: 

231 self.sprints_dir = Path(".sprints") 

232 self.sprints_dir.mkdir(parents=True, exist_ok=True) 

233 

234 def get_default_options(self) -> SprintOptions: 

235 """Get default SprintOptions from config or hardcoded defaults. 

236 

237 Returns: 

238 SprintOptions with values from config if available, else defaults 

239 """ 

240 if self.config is not None: 

241 return SprintOptions( 

242 timeout=self.config.sprints.default_timeout, 

243 max_workers=self.config.sprints.default_max_workers, 

244 ) 

245 return SprintOptions() 

246 

247 def create( 

248 self, 

249 name: str, 

250 issues: list[str], 

251 description: str = "", 

252 options: SprintOptions | None = None, 

253 ) -> Sprint: 

254 """Create a new sprint. 

255 

256 Args: 

257 name: Sprint identifier 

258 issues: List of issue IDs 

259 description: Human-readable description 

260 options: Optional execution options 

261 

262 Returns: 

263 Created Sprint instance 

264 """ 

265 sprint = Sprint( 

266 name=name, 

267 description=description, 

268 issues=[i.strip().upper() for i in issues], 

269 created=datetime.now(UTC).isoformat(), 

270 options=options, 

271 ) 

272 sprint.save(self.sprints_dir) 

273 return sprint 

274 

275 def load(self, name: str) -> Sprint | None: 

276 """Load a sprint by name. 

277 

278 Args: 

279 name: Sprint name 

280 

281 Returns: 

282 Sprint instance or None if not found 

283 """ 

284 return Sprint.load(self.sprints_dir, name) 

285 

286 def load_or_resolve(self, arg: str) -> "Sprint | None": 

287 """Load a sprint by name or resolve an EPIC ID to an ephemeral Sprint. 

288 

289 If `arg` matches ^EPIC-\\d+$ (case-insensitive), resolves the EPIC's 

290 active children via union of forward (relates_to:) and backward (parent:) 

291 lookups, filtered to active statuses and ordered by dependency graph. 

292 Otherwise falls through to the file-based load() path. 

293 

294 Args: 

295 arg: Sprint name (file-based) or EPIC ID matching ^EPIC-\\d+$ 

296 

297 Returns: 

298 Sprint instance, or None if not found / EPIC not found 

299 """ 

300 if not _EPIC_ID_RE.match(arg): 

301 return self.load(arg) 

302 

303 epic_id = arg.upper() 

304 

305 if not self.config: 

306 return self.load(arg) 

307 

308 epic_path = self._find_issue_path(epic_id) 

309 if epic_path is None: 

310 return None 

311 

312 from little_loops.issue_parser import IssueParser, find_issues 

313 

314 parser = IssueParser(self.config) 

315 try: 

316 epic_info = parser.parse_file(epic_path) 

317 except Exception as e: 

318 logger.warning("Failed to parse EPIC file %s: %s", epic_path, e) 

319 return None 

320 

321 # Forward lookup: relates_to on the EPIC file 

322 forward_ids: set[str] = set(epic_info.relates_to) 

323 

324 # Backward lookup: scan all active issues for parent == epic_id 

325 all_active = find_issues(self.config, status_filter=_ACTIVE_STATUSES) 

326 backward_ids = {info.issue_id for info in all_active if info.parent == epic_id} 

327 

328 # Union + dedup; intersect with active set so forward refs to done issues are dropped 

329 active_ids_set = {info.issue_id for info in all_active} 

330 child_ids = (forward_ids | backward_ids) & active_ids_set 

331 child_infos = [info for info in all_active if info.issue_id in child_ids] 

332 

333 sprint_name = f"epic-{epic_id.split('-', 1)[1]}" 

334 

335 if not child_infos: 

336 logger.info("EPIC %s has no active children", epic_id) 

337 return Sprint( 

338 name=sprint_name, 

339 description=f"Resolved from {epic_id}", 

340 issues=[], 

341 created=datetime.now(UTC).isoformat(), 

342 ) 

343 

344 from little_loops.dependency_graph import DependencyGraph 

345 

346 dep_graph = DependencyGraph.from_issues(child_infos, all_known_ids=active_ids_set) 

347 try: 

348 waves = dep_graph.get_execution_waves() 

349 ordered_ids = [issue.issue_id for wave in waves for issue in wave] 

350 except ValueError: 

351 ordered_ids = [ 

352 info.issue_id 

353 for info in sorted(child_infos, key=lambda i: (i.priority or "P5", i.issue_id)) 

354 ] 

355 

356 return Sprint( 

357 name=sprint_name, 

358 description=f"Resolved from {epic_id}: {epic_info.title}", 

359 issues=ordered_ids, 

360 created=datetime.now(UTC).isoformat(), 

361 ) 

362 

363 def list_all(self) -> list[Sprint]: 

364 """List all sprints. 

365 

366 Returns: 

367 List of Sprint instances, sorted by name 

368 """ 

369 sprints = [] 

370 for path in sorted(self.sprints_dir.glob("*.yaml")): 

371 sprint = Sprint.load(self.sprints_dir, path.stem) 

372 if sprint: 

373 sprints.append(sprint) 

374 return sprints 

375 

376 def delete(self, name: str) -> bool: 

377 """Delete a sprint. 

378 

379 Args: 

380 name: Sprint name 

381 

382 Returns: 

383 True if deleted, False if not found 

384 """ 

385 sprint_path = self.sprints_dir / f"{name}.yaml" 

386 if not sprint_path.exists(): 

387 return False 

388 sprint_path.unlink() 

389 return True 

390 

391 def _find_issue_path(self, issue_id: str) -> Path | None: 

392 """Find the filesystem path for an issue ID. 

393 

394 Searches all configured issue categories for a file matching the issue ID. 

395 

396 Args: 

397 issue_id: Issue ID to locate (e.g. "BUG-001") 

398 

399 Returns: 

400 Path to the issue file, or None if not found 

401 """ 

402 if not self.config: 

403 return None 

404 for category in self.config.issue_categories: 

405 issue_dir = self.config.get_issue_dir(category) 

406 for path in issue_dir.glob(f"*-{issue_id}-*.md"): 

407 return path 

408 return None 

409 

410 def validate_issues(self, issues: list[str]) -> dict[str, Path]: 

411 """Validate that issue IDs exist. 

412 

413 Args: 

414 issues: List of issue IDs to validate 

415 

416 Returns: 

417 Dictionary mapping valid issue IDs to their file paths 

418 """ 

419 if not self.config: 

420 # No config provided, skip validation 

421 return {} 

422 

423 valid = {} 

424 for issue_id in issues: 

425 path = self._find_issue_path(issue_id) 

426 if path is not None: 

427 valid[issue_id] = path 

428 return valid 

429 

430 def load_issue_infos(self, issues: list[str]) -> list["IssueInfo"]: 

431 """Load IssueInfo objects for the given issue IDs. 

432 

433 Args: 

434 issues: List of issue IDs to load 

435 

436 Returns: 

437 List of IssueInfo objects (only for issues that exist) 

438 """ 

439 from little_loops.issue_parser import IssueParser 

440 

441 if not self.config: 

442 return [] 

443 

444 parser = IssueParser(self.config) 

445 result: list[IssueInfo] = [] 

446 for issue_id in issues: 

447 path = self._find_issue_path(issue_id) 

448 if path is not None: 

449 try: 

450 info = parser.parse_file(path) 

451 result.append(info) 

452 except Exception as e: 

453 logger.warning("Failed to parse issue file %s: %s", path, e) 

454 return result