Coverage for little_loops / cli / sprint / run.py: 8%

394 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2026-06-26 17:38 -0500

1"""ll-sprint run subcommand with signal handling and state management.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import signal 

7import subprocess 

8import sys 

9import time 

10from pathlib import Path 

11from types import FrameType 

12from typing import TYPE_CHECKING 

13 

14from little_loops.cli.output import use_color_enabled 

15from little_loops.cli.sprint._helpers import _build_issue_contents, _render_dependency_analysis 

16from little_loops.cli_args import _id_matches, parse_issue_ids, parse_issue_types, parse_labels 

17from little_loops.dependency_graph import DependencyGraph, refine_waves_for_contention 

18from little_loops.issue_manager import IssueProcessingResult, process_issue_inplace 

19from little_loops.logger import Logger, format_duration 

20from little_loops.parallel.orchestrator import ParallelOrchestrator 

21from little_loops.parallel.types import SprintWorkerContext 

22from little_loops.sprint import SprintManager, SprintState 

23 

24if TYPE_CHECKING: 

25 import argparse 

26 

27 from little_loops.config import BRConfig 

28 from little_loops.issue_parser import IssueInfo 

29 

30# Module-level shutdown flag for ll-sprint signal handling (ENH-183) 

31_sprint_shutdown_requested: bool = False 

32_sprint_logger: Logger | None = None 

33 

34 

35class IssueWallClockTimeout(Exception): 

36 """Raised when an issue exceeds its per-issue wall-clock time budget (BUG-2144).""" 

37 

38 def __init__(self, issue_id: str) -> None: 

39 self.issue_id = issue_id 

40 super().__init__(f"{issue_id}: wall-clock timeout") 

41 

42 

43def _run_issue_with_wall_clock_timeout( 

44 issue: IssueInfo, 

45 config: BRConfig, 

46 logger: Logger, 

47 dry_run: bool, 

48 max_seconds: int, 

49 sprint_context: SprintWorkerContext | None = None, 

50) -> IssueProcessingResult: 

51 """Wrap process_issue_inplace with a SIGALRM-based wall-clock timeout. 

52 

53 If the issue takes longer than max_seconds (across all subprocess spawns, 

54 including Option J continuations), SIGALRM fires, IssueWallClockTimeout is 

55 raised, and a failed IssueProcessingResult with failure_reason=WALL_CLOCK_TIMEOUT 

56 is returned. The alarm is always cleared in the finally block. 

57 """ 

58 

59 def _timeout_handler(signum: int, frame: FrameType | None) -> None: 

60 raise IssueWallClockTimeout(issue.issue_id) 

61 

62 prev_handler = signal.signal(signal.SIGALRM, _timeout_handler) 

63 signal.alarm(max_seconds) 

64 issue_start = time.monotonic() 

65 try: 

66 return process_issue_inplace( 

67 info=issue, 

68 config=config, 

69 logger=logger, 

70 dry_run=dry_run, 

71 sprint_context=sprint_context, 

72 ) 

73 except IssueWallClockTimeout: 

74 elapsed = time.monotonic() - issue_start 

75 logger.error( 

76 f" {issue.issue_id}: wall-clock timeout after {int(elapsed)}s " 

77 f"(limit {max_seconds}s) — marking failed" 

78 ) 

79 return IssueProcessingResult( 

80 success=False, 

81 duration=elapsed, 

82 issue_id=issue.issue_id, 

83 failure_reason="WALL_CLOCK_TIMEOUT", 

84 ) 

85 finally: 

86 signal.alarm(0) 

87 signal.signal(signal.SIGALRM, prev_handler) 

88 

89 

90def _detect_current_branch() -> str: 

91 """Return the current git branch name, defaulting to 'main' on failure.""" 

92 result = subprocess.run( 

93 ["git", "rev-parse", "--abbrev-ref", "HEAD"], 

94 capture_output=True, 

95 text=True, 

96 ) 

97 return result.stdout.strip() if result.returncode == 0 else "main" 

98 

99 

100def _sprint_signal_handler(signum: int, frame: FrameType | None) -> None: 

101 """Handle shutdown signals gracefully for ll-sprint. 

102 

103 First signal: Set shutdown flag for graceful exit after current wave. 

104 Second signal: Force immediate exit. 

105 """ 

106 global _sprint_shutdown_requested 

107 if _sprint_shutdown_requested: 

108 # Second signal - force exit 

109 msg = "\nForce shutdown requested" 

110 if _sprint_logger is not None: 

111 _sprint_logger.warning(msg) 

112 else: 

113 print(msg, file=sys.stderr) 

114 sys.exit(1) 

115 _sprint_shutdown_requested = True 

116 msg = "\nShutdown requested, will exit after current wave..." 

117 if _sprint_logger is not None: 

118 _sprint_logger.warning(msg) 

119 else: 

120 print(msg, file=sys.stderr) 

121 

122 

123def _get_sprint_state_file() -> Path: 

124 """Get path to sprint state file.""" 

125 return Path.cwd() / ".sprint-state.json" 

126 

127 

128def _load_sprint_state(logger: Logger) -> SprintState | None: 

129 """Load sprint state from file.""" 

130 import json 

131 

132 state_file = _get_sprint_state_file() 

133 if not state_file.exists(): 

134 return None 

135 try: 

136 data = json.loads(state_file.read_text()) 

137 state = SprintState.from_dict(data) 

138 logger.info(f"State loaded from {state_file}") 

139 return state 

140 except (json.JSONDecodeError, KeyError) as e: 

141 logger.warning(f"Failed to load state: {e}") 

142 return None 

143 

144 

145def _save_sprint_state(state: SprintState, logger: Logger) -> None: 

146 """Save sprint state to file.""" 

147 import json 

148 from datetime import datetime 

149 

150 state.last_checkpoint = datetime.now().isoformat() 

151 state_file = _get_sprint_state_file() 

152 state_file.write_text(json.dumps(state.to_dict(), indent=2)) 

153 logger.info(f"State saved to {state_file}") 

154 

155 

156def _cleanup_sprint_state(logger: Logger) -> None: 

157 """Remove sprint state file.""" 

158 state_file = _get_sprint_state_file() 

159 if state_file.exists(): 

160 state_file.unlink() 

161 logger.info("Sprint state file cleaned up") 

162 

163 

164def _run_learning_gate_preflight( 

165 args: argparse.Namespace, 

166 issue_infos: list[IssueInfo], 

167 config: BRConfig, 

168 logger: Logger, 

169) -> int: 

170 """Run the learning-test pre-flight gate before sprint execution (ENH-2210). 

171 

172 Aggregates all learning_tests_required targets across sprint issues, 

173 deduplicates them (preserving first-occurrence order), and runs 

174 ready-to-implement-gate once. If any target is unproven the sprint aborts 

175 with a message identifying the blocking targets and the issue IDs that 

176 contributed each one. 

177 

178 Returns: 

179 0 if the gate passes or is skipped; 1 if the gate blocks the sprint. 

180 """ 

181 if not config.learning_tests.enabled: 

182 return 0 

183 

184 if getattr(args, "skip_learning_gate", False): 

185 logger.info("Learning gate pre-flight skipped (--skip-learning-gate)") 

186 return 0 

187 

188 from little_loops.learning_tests.extractor import extract_learning_targets 

189 

190 # Map target → contributing issue IDs (for attribution in error output) 

191 target_to_issues: dict[str, list[str]] = {} 

192 seen: set[str] = set() 

193 all_targets: list[str] = [] 

194 

195 for info in issue_infos: 

196 if info.learning_tests_required is not None: 

197 targets = info.learning_tests_required 

198 else: 

199 # TODO(stale-after-ENH-2209): remove fallback once all active sprint 

200 # issues have been re-refined with ENH-2209 auto-population. 

201 try: 

202 targets = extract_learning_targets(info.path.read_text()) 

203 except OSError: 

204 targets = [] 

205 

206 for target in targets: 

207 if target not in seen: 

208 seen.add(target) 

209 all_targets.append(target) 

210 target_to_issues.setdefault(target, []).append(info.issue_id) 

211 

212 if not all_targets: 

213 return 0 

214 

215 logger.info( 

216 f"Learning gate pre-flight: checking {len(all_targets)} target(s): {', '.join(all_targets)}" 

217 ) 

218 

219 cmd = [ 

220 "ll-loop", 

221 "run", 

222 "ready-to-implement-gate", 

223 "--context", 

224 f"targets={','.join(all_targets)}", 

225 ] 

226 result = subprocess.run(cmd, capture_output=True, text=True) 

227 

228 if result.returncode != 0: 

229 logger.error("Learning gate pre-flight FAILED — sprint aborted") 

230 for target in all_targets: 

231 contributing = ", ".join(target_to_issues.get(target, [])) 

232 logger.error(f" Blocking target '{target}': required by {contributing}") 

233 return 1 

234 

235 logger.info("Learning gate pre-flight: all targets proven") 

236 return 0 

237 

238 

239def _cmd_sprint_run( 

240 args: argparse.Namespace, 

241 manager: SprintManager, 

242 config: BRConfig, 

243) -> int: 

244 """Execute a sprint with dependency-aware scheduling.""" 

245 from datetime import datetime 

246 

247 global _sprint_logger 

248 logger = Logger(verbose=not args.quiet, use_color=use_color_enabled()) 

249 _sprint_logger = logger 

250 

251 # Setup signal handlers for graceful shutdown (ENH-183) 

252 global _sprint_shutdown_requested 

253 _sprint_shutdown_requested = False # Reset in case of multiple runs 

254 signal.signal(signal.SIGINT, _sprint_signal_handler) 

255 signal.signal(signal.SIGTERM, _sprint_signal_handler) 

256 

257 handoff_threshold = getattr(args, "handoff_threshold", None) 

258 if handoff_threshold is not None: 

259 os.environ["LL_HANDOFF_THRESHOLD"] = str(handoff_threshold) 

260 

261 context_limit = getattr(args, "context_limit", None) 

262 if context_limit is not None: 

263 os.environ["LL_CONTEXT_LIMIT"] = str(context_limit) 

264 

265 sprint = manager.load_or_resolve(args.sprint) 

266 if not sprint: 

267 logger.error(f"Sprint not found: {args.sprint}") 

268 return 1 

269 

270 if not sprint.issues: 

271 logger.info(f"Sprint '{sprint.name}' has no active issues — nothing to process") 

272 return 0 

273 

274 if getattr(args, "save", False): 

275 saved_path = sprint.save(manager.sprints_dir) 

276 logger.info(f"Sprint saved to {saved_path}") 

277 

278 # Apply skip filter if provided 

279 issues_to_process = list(sprint.issues) 

280 skip_ids = parse_issue_ids(args.skip) 

281 if skip_ids: 

282 original_count = len(issues_to_process) 

283 issues_to_process = [i for i in issues_to_process if i not in skip_ids] 

284 skipped = original_count - len(issues_to_process) 

285 if skipped > 0: 

286 logger.info(f"Skipping {skipped} issue(s): {', '.join(sorted(skip_ids))}") 

287 

288 # Apply only filter if provided 

289 only_ids = parse_issue_ids(getattr(args, "only", None)) 

290 if only_ids: 

291 invalid_only = {p for p in only_ids if not any(_id_matches(i, p) for i in sprint.issues)} 

292 if invalid_only: 

293 logger.error( 

294 f"Issue(s) not found in sprint definition: {', '.join(sorted(invalid_only))}" 

295 ) 

296 return 1 

297 issues_to_process = [ 

298 i for i in issues_to_process if any(_id_matches(i, p) for p in only_ids) 

299 ] 

300 logger.info( 

301 f"Processing only {len(issues_to_process)} issue(s): {', '.join(sorted(only_ids))}" 

302 ) 

303 

304 # Apply type filter if provided 

305 type_prefixes = parse_issue_types(getattr(args, "type", None)) 

306 if type_prefixes: 

307 original_count = len(issues_to_process) 

308 issues_to_process = [i for i in issues_to_process if i.split("-", 1)[0] in type_prefixes] 

309 filtered = original_count - len(issues_to_process) 

310 if filtered > 0: 

311 logger.info(f"Filtered {filtered} issue(s) by type: {', '.join(sorted(type_prefixes))}") 

312 

313 # Pre-validate: skip issues with status: done/cancelled (ENH-1424) 

314 pre_completed_skipped: list[str] = [] 

315 

316 # Validate issues exist 

317 valid = manager.validate_issues(issues_to_process) 

318 invalid = set(issues_to_process) - set(valid.keys()) 

319 

320 if invalid: 

321 logger.error(f"Issue IDs not found: {', '.join(sorted(invalid))}") 

322 logger.info("Cannot execute sprint with missing issues") 

323 return 1 

324 

325 # Write milestone: field back to issue files for all valid issues in this sprint 

326 from little_loops.frontmatter import parse_frontmatter, update_frontmatter 

327 

328 for _issue_id, path in valid.items(): 

329 content = path.read_text(encoding="utf-8") 

330 updated = update_frontmatter(content, {"milestone": sprint.name}) 

331 if updated != content: 

332 path.write_text(updated, encoding="utf-8") 

333 logger.info(f"Set milestone: {sprint.name!r} on {len(valid)} issue file(s)") 

334 

335 # Skip issues already completed via frontmatter status (ENH-1424) 

336 still_active: list[str] = [] 

337 for issue_id in issues_to_process: 

338 path = valid[issue_id] 

339 fm = parse_frontmatter(path.read_text(encoding="utf-8")) 

340 if fm.get("status", "open") in ("done", "cancelled"): 

341 logger.info(f" {issue_id}: already completed (status: {fm.get('status')}), skipping") 

342 pre_completed_skipped.append(issue_id) 

343 else: 

344 still_active.append(issue_id) 

345 if pre_completed_skipped: 

346 logger.info( 

347 f"Pre-validation: {len(pre_completed_skipped)} issue(s) already completed, " 

348 f"{len(still_active)} active" 

349 ) 

350 issues_to_process = still_active 

351 

352 if pre_completed_skipped and not issues_to_process: 

353 logger.info("All sprint issues already completed - nothing to process") 

354 return 0 

355 

356 # Load full IssueInfo objects for dependency analysis 

357 issue_infos = manager.load_issue_infos(issues_to_process) 

358 if not issue_infos: 

359 logger.error("No issue files found") 

360 return 1 

361 

362 # Apply label filter if provided (must run after IssueInfo load since labels come from frontmatter) 

363 label_filter = parse_labels(getattr(args, "label", None)) 

364 if label_filter: 

365 original_count = len(issue_infos) 

366 issue_infos = [i for i in issue_infos if any(lb.lower() in label_filter for lb in i.labels)] 

367 filtered = original_count - len(issue_infos) 

368 if filtered > 0: 

369 logger.info(f"Filtered {filtered} issue(s) by label: {', '.join(sorted(label_filter))}") 

370 issues_to_process = [i.issue_id for i in issue_infos] 

371 

372 # Learning-test pre-flight gate (ENH-2210) 

373 preflight_result = _run_learning_gate_preflight(args, issue_infos, config, logger) 

374 if preflight_result != 0: 

375 return preflight_result 

376 

377 # Gather all issue IDs on disk to avoid false "nonexistent" warnings 

378 from little_loops.dependency_mapper import gather_all_issue_ids 

379 

380 issues_dir = config.project_root / config.issues.base_dir 

381 all_known_ids = gather_all_issue_ids(issues_dir, config=config) 

382 

383 # Dependency analysis (ENH-301) 

384 if not getattr(args, "skip_analysis", False): 

385 from little_loops.dependency_mapper import analyze_dependencies 

386 

387 issue_contents = _build_issue_contents(issue_infos) 

388 dep_config = config.dependency_mapping 

389 dep_report = analyze_dependencies( 

390 issue_infos, issue_contents, all_known_ids=all_known_ids, config=dep_config 

391 ) 

392 _render_dependency_analysis(dep_report, logger, config=dep_config) 

393 

394 # Build dependency graph 

395 dep_graph = DependencyGraph.from_issues(issue_infos, all_known_ids=all_known_ids) 

396 

397 # Detect cycles 

398 if dep_graph.has_cycles(): 

399 cycles = dep_graph.detect_cycles() 

400 for cycle in cycles: 

401 logger.error(f"Dependency cycle detected: {' -> '.join(cycle)}") 

402 return 1 

403 

404 # Get execution waves 

405 try: 

406 waves = dep_graph.get_execution_waves() 

407 except ValueError as e: 

408 logger.error(str(e)) 

409 return 1 

410 

411 # Refine waves for file overlap (ENH-306) 

412 waves, contention_notes = refine_waves_for_contention(waves, config=config.dependency_mapping) 

413 

414 # Display execution plan 

415 logger.info(f"Running sprint: {sprint.name}") 

416 logger.info("Dependency analysis:") 

417 for i, wave in enumerate(waves, 1): 

418 issue_ids = ", ".join(issue.issue_id for issue in wave) 

419 note = contention_notes[i - 1] if contention_notes else None 

420 if note: 

421 logger.info( 

422 f" Wave {i}: {issue_ids}" 

423 f" [sub-wave {note.sub_wave_index + 1}/{note.total_sub_waves}]" 

424 ) 

425 else: 

426 logger.info(f" Wave {i}: {issue_ids}") 

427 

428 # Warn when issues were serialized conservatively due to missing file hints (BUG-2142) 

429 if any(n is not None and getattr(n, "has_unknown_hints", False) for n in contention_notes): 

430 logger.warning( 

431 "Some issues lack '### Files to Modify' sections — serialized conservatively." 

432 " Add file hints for more accurate wave scheduling." 

433 ) 

434 

435 if args.dry_run: 

436 logger.info("\nDry run mode - no changes will be made") 

437 return 0 

438 

439 # Initialize or load state 

440 state: SprintState 

441 start_wave = 1 

442 

443 if args.resume: 

444 loaded_state = _load_sprint_state(logger) 

445 if loaded_state and loaded_state.sprint_name == sprint.name: 

446 state = loaded_state 

447 # Find first incomplete wave by checking completed issues 

448 completed_set = set(state.completed_issues) 

449 for i, wave in enumerate(waves, 1): 

450 wave_issue_ids = {issue.issue_id for issue in wave} 

451 if not wave_issue_ids.issubset(completed_set): 

452 start_wave = i 

453 break 

454 else: 

455 # All waves completed 

456 logger.info("Sprint already completed - nothing to resume") 

457 _cleanup_sprint_state(logger) 

458 return 0 

459 logger.info(f"Resuming from wave {start_wave}/{len(waves)}") 

460 logger.info(f" Previously completed: {len(state.completed_issues)} issues") 

461 else: 

462 if loaded_state: 

463 logger.warning( 

464 f"State file is for sprint '{loaded_state.sprint_name}', " 

465 f"not '{sprint.name}' - starting fresh" 

466 ) 

467 else: 

468 logger.warning("No valid state found - starting fresh") 

469 state = SprintState( 

470 sprint_name=sprint.name, 

471 started_at=datetime.now().isoformat(), 

472 ) 

473 else: 

474 # Fresh start - delete any old state 

475 _cleanup_sprint_state(logger) 

476 state = SprintState( 

477 sprint_name=sprint.name, 

478 started_at=datetime.now().isoformat(), 

479 ) 

480 

481 # Track exit status for error handling (ENH-185) 

482 exit_code = 0 

483 

484 try: 

485 # Determine max workers 

486 max_workers = args.max_workers or (sprint.options.max_workers if sprint.options else 2) 

487 

488 # Execute wave by wave 

489 completed: set[str] = set(state.completed_issues) 

490 failed_waves = 0 

491 total_duration = 0.0 

492 total_waves = len(waves) 

493 _fb_warning_emitted = False # deduplicate: fire once per sprint run (ENH-2176) 

494 

495 for wave_num, wave in enumerate(waves, 1): 

496 # Check for shutdown request (ENH-183) 

497 if _sprint_shutdown_requested: 

498 logger.warning("Shutdown requested - saving state and exiting") 

499 _save_sprint_state(state, logger) 

500 exit_code = 1 

501 return exit_code 

502 

503 # Skip already-completed waves when resuming 

504 if wave_num < start_wave: 

505 continue 

506 

507 wave_ids = [issue.issue_id for issue in wave] 

508 state.current_wave = wave_num 

509 logger.info(f"\nProcessing wave {wave_num}/{total_waves}: {', '.join(wave_ids)}") 

510 

511 wave_note = ( 

512 contention_notes[wave_num - 1] 

513 if contention_notes and wave_num - 1 < len(contention_notes) 

514 else None 

515 ) 

516 is_contention_subwave = wave_note is not None 

517 

518 if len(wave) == 1 or is_contention_subwave: 

519 # Single issue OR contention sub-wave — process in-place sequentially 

520 # (contention sub-waves are displayed as "serialized steps" so must run that way) 

521 max_wall_clock = config.sprints.max_issue_wall_clock_time 

522 

523 wave_failed = False 

524 _current_branch = _detect_current_branch() 

525 _feature_branches_arg = getattr(args, "feature_branches", None) 

526 effective_feature_branches = ( 

527 _feature_branches_arg 

528 if _feature_branches_arg is not None 

529 else config.parallel.use_feature_branches 

530 ) 

531 if effective_feature_branches and not _fb_warning_emitted: 

532 logger.warning( 

533 "feature-branch mode does not apply to single-issue / contention" 

534 f" sub-waves; these issues run in-place on '{_current_branch}'" 

535 ) 

536 _fb_warning_emitted = True 

537 for issue in wave: 

538 # TODO(ENH-1686): sprint sequential path not yet live-written 

539 issue_result = _run_issue_with_wall_clock_timeout( 

540 issue=issue, 

541 config=config, 

542 logger=logger, 

543 dry_run=args.dry_run, 

544 max_seconds=max_wall_clock, 

545 sprint_context=SprintWorkerContext( 

546 issue_id=issue.issue_id, 

547 branch=_current_branch, 

548 ), 

549 ) 

550 total_duration += issue_result.duration 

551 if issue_result.success: 

552 completed.add(issue.issue_id) 

553 state.completed_issues.append(issue.issue_id) 

554 state.timing[issue.issue_id] = {"total": issue_result.duration} 

555 logger.success(f" {issue.issue_id}: completed") 

556 elif issue_result.was_blocked: 

557 completed.add(issue.issue_id) 

558 state.skipped_blocked_issues[issue.issue_id] = issue_result.failure_reason 

559 logger.warning(f" {issue.issue_id}: skipped (blocked by open dependency)") 

560 else: 

561 wave_failed = True 

562 completed.add(issue.issue_id) 

563 state.failed_issues[issue.issue_id] = ( 

564 issue_result.failure_reason or "Issue processing failed" 

565 ) 

566 logger.warning(f" {issue.issue_id}: failed") 

567 if wave_failed: 

568 failed_waves += 1 

569 logger.warning(f"Wave {wave_num}/{total_waves} had failures") 

570 else: 

571 logger.success( 

572 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}" 

573 ) 

574 _save_sprint_state(state, logger) 

575 if wave_num < total_waves: 

576 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...") 

577 # Check for shutdown before next wave (ENH-183) 

578 if _sprint_shutdown_requested: 

579 logger.warning("Shutdown requested - exiting after wave completion") 

580 exit_code = 1 

581 return exit_code 

582 else: 

583 # Multi-issue — use ParallelOrchestrator with worktrees 

584 only_ids = set(wave_ids) 

585 # Detect current branch for rebase/merge operations (BUG-439) 

586 _br = subprocess.run( 

587 ["git", "rev-parse", "--abbrev-ref", "HEAD"], 

588 capture_output=True, 

589 text=True, 

590 cwd=Path.cwd(), 

591 ) 

592 _base_branch = _br.stdout.strip() if _br.returncode == 0 else "main" 

593 # Runtime overlap detection disabled for sprints (ENH-512): 

594 # refine_waves_for_contention() already splits overlapping 

595 # issues into separate sub-waves before dispatch. 

596 parallel_config = config.create_parallel_config( 

597 max_workers=min(max_workers, len(wave)), 

598 only_ids=only_ids, 

599 dry_run=args.dry_run, 

600 overlap_detection=False, 

601 serialize_overlapping=True, 

602 base_branch=_base_branch, 

603 clean_start=True, # Sprint manages its own state; don't load stale orchestrator state 

604 use_feature_branches=getattr(args, "feature_branches", None), 

605 ) 

606 

607 from little_loops.events import EventBus 

608 from little_loops.extension import wire_extensions 

609 from little_loops.transport import wire_transports 

610 

611 event_bus = EventBus() 

612 wire_extensions(event_bus, config.extensions) 

613 wire_transports(event_bus, config.events) 

614 orchestrator = ParallelOrchestrator( 

615 parallel_config, 

616 config, 

617 Path.cwd(), 

618 wave_label=f"Wave {wave_num}/{total_waves}", 

619 event_bus=event_bus, 

620 ) 

621 result = orchestrator.run() 

622 total_duration += orchestrator.execution_duration 

623 

624 # Track completed/failed from this wave using per-issue results 

625 actually_completed = set(orchestrator.queue.completed_ids) 

626 actually_failed = set(orchestrator.queue.failed_ids) 

627 

628 for issue_id in wave_ids: 

629 if issue_id in actually_completed: 

630 completed.add(issue_id) 

631 state.completed_issues.append(issue_id) 

632 state.timing[issue_id] = { 

633 "total": orchestrator.execution_duration / len(wave) 

634 } 

635 elif issue_id in actually_failed: 

636 completed.add(issue_id) 

637 state.failed_issues[issue_id] = "Issue failed during wave execution" 

638 # else: issue was neither completed nor failed (interrupted/stranded) 

639 # — leave untracked so it can be retried on resume 

640 

641 # Sequential retry for failed issues (ENH-308) 

642 if actually_failed: 

643 logger.info(f"Retrying {len(actually_failed)} failed issue(s) sequentially...") 

644 retried_ok = 0 

645 for issue in wave: 

646 if issue.issue_id not in actually_failed: 

647 continue 

648 logger.info(f" Retrying {issue.issue_id} in-place...") 

649 # TODO(ENH-1686): sprint sequential path not yet live-written 

650 retry_result = process_issue_inplace( 

651 info=issue, 

652 config=config, 

653 logger=logger, 

654 dry_run=args.dry_run, 

655 sprint_context=SprintWorkerContext( 

656 issue_id=issue.issue_id, 

657 branch=_detect_current_branch(), 

658 ), 

659 ) 

660 total_duration += retry_result.duration 

661 if retry_result.success: 

662 retried_ok += 1 

663 state.failed_issues.pop(issue.issue_id, None) 

664 state.completed_issues.append(issue.issue_id) 

665 state.timing[issue.issue_id] = {"total": retry_result.duration} 

666 logger.success(f" Retry succeeded: {issue.issue_id}") 

667 elif retry_result.was_blocked: 

668 state.failed_issues.pop(issue.issue_id, None) 

669 state.skipped_blocked_issues[issue.issue_id] = ( 

670 retry_result.failure_reason 

671 ) 

672 logger.warning( 

673 f" Retry skipped: {issue.issue_id} (blocked by open dependency)" 

674 ) 

675 else: 

676 logger.warning(f" Retry failed: {issue.issue_id}") 

677 if retried_ok > 0: 

678 logger.info( 

679 f"Sequential retry recovered {retried_ok}/{len(actually_failed)} issue(s)" 

680 ) 

681 

682 # Check whether failures remain after retry (ENH-308) 

683 remaining_failures = {iid for iid in actually_failed if iid in state.failed_issues} 

684 if result == 0 or not remaining_failures: 

685 logger.success( 

686 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}" 

687 ) 

688 else: 

689 failed_waves += 1 

690 logger.warning(f"Wave {wave_num}/{total_waves} had failures") 

691 _save_sprint_state(state, logger) 

692 if wave_num < total_waves: 

693 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...") 

694 # Check for shutdown before next wave (ENH-183) 

695 if _sprint_shutdown_requested: 

696 logger.warning("Shutdown requested - exiting after wave completion") 

697 exit_code = 1 

698 return exit_code 

699 

700 wave_word = "wave" if len(waves) == 1 else "waves" 

701 skip_msg = ( 

702 f", {len(pre_completed_skipped)} already completed (skipped)" 

703 if pre_completed_skipped 

704 else "" 

705 ) 

706 blocked_msg = ( 

707 f", {len(state.skipped_blocked_issues)} skipped (blocked)" 

708 if state.skipped_blocked_issues 

709 else "" 

710 ) 

711 logger.info( 

712 f"\nSprint completed: {len(completed)} issues processed " 

713 f"({len(waves)} {wave_word}){skip_msg}{blocked_msg}" 

714 ) 

715 logger.timing(f"Total execution time: {format_duration(total_duration)}") 

716 if failed_waves > 0: 

717 logger.warning(f"{failed_waves} wave(s) had failures") 

718 exit_code = 1 

719 else: 

720 # Clean up state on successful completion 

721 _cleanup_sprint_state(logger) 

722 

723 except KeyboardInterrupt: 

724 # Belt-and-suspenders with signal handler (ENH-185) 

725 logger.warning("Sprint interrupted by user (KeyboardInterrupt)") 

726 exit_code = 130 

727 

728 except Exception as e: 

729 # Catch unexpected exceptions (ENH-185) 

730 logger.error(f"Sprint failed unexpectedly: {e}") 

731 exit_code = 1 

732 

733 finally: 

734 # Guaranteed state save on any non-success exit (ENH-185) 

735 if exit_code != 0: 

736 _save_sprint_state(state, logger) 

737 logger.info("State saved before exit") 

738 

739 _sprint_logger = None # Clear for test isolation 

740 return exit_code