Coverage for little_loops / cli / sprint / run.py: 7%
292 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""ll-sprint run subcommand with signal handling and state management."""
3from __future__ import annotations
5import os
6import signal
7import subprocess
8import sys
9from pathlib import Path
10from types import FrameType
11from typing import TYPE_CHECKING
13from little_loops.cli.sprint._helpers import _build_issue_contents, _render_dependency_analysis
14from little_loops.cli_args import _id_matches, parse_issue_ids, parse_issue_types
15from little_loops.dependency_graph import DependencyGraph, refine_waves_for_contention
16from little_loops.logger import Logger, format_duration
17from little_loops.parallel.orchestrator import ParallelOrchestrator
18from little_loops.sprint import SprintManager, SprintState
20if TYPE_CHECKING:
21 import argparse
23 from little_loops.config import BRConfig
25# Module-level shutdown flag for ll-sprint signal handling (ENH-183)
26_sprint_shutdown_requested: bool = False
29def _sprint_signal_handler(signum: int, frame: FrameType | None) -> None:
30 """Handle shutdown signals gracefully for ll-sprint.
32 First signal: Set shutdown flag for graceful exit after current wave.
33 Second signal: Force immediate exit.
34 """
35 global _sprint_shutdown_requested
36 if _sprint_shutdown_requested:
37 # Second signal - force exit
38 print("\nForce shutdown requested", file=sys.stderr)
39 sys.exit(1)
40 _sprint_shutdown_requested = True
41 print("\nShutdown requested, will exit after current wave...", file=sys.stderr)
44def _get_sprint_state_file() -> Path:
45 """Get path to sprint state file."""
46 return Path.cwd() / ".sprint-state.json"
49def _load_sprint_state(logger: Logger) -> SprintState | None:
50 """Load sprint state from file."""
51 import json
53 state_file = _get_sprint_state_file()
54 if not state_file.exists():
55 return None
56 try:
57 data = json.loads(state_file.read_text())
58 state = SprintState.from_dict(data)
59 logger.info(f"State loaded from {state_file}")
60 return state
61 except (json.JSONDecodeError, KeyError) as e:
62 logger.warning(f"Failed to load state: {e}")
63 return None
66def _save_sprint_state(state: SprintState, logger: Logger) -> None:
67 """Save sprint state to file."""
68 import json
69 from datetime import datetime
71 state.last_checkpoint = datetime.now().isoformat()
72 state_file = _get_sprint_state_file()
73 state_file.write_text(json.dumps(state.to_dict(), indent=2))
74 logger.info(f"State saved to {state_file}")
77def _cleanup_sprint_state(logger: Logger) -> None:
78 """Remove sprint state file."""
79 state_file = _get_sprint_state_file()
80 if state_file.exists():
81 state_file.unlink()
82 logger.info("Sprint state file cleaned up")
85def _cmd_sprint_run(
86 args: argparse.Namespace,
87 manager: SprintManager,
88 config: BRConfig,
89) -> int:
90 """Execute a sprint with dependency-aware scheduling."""
91 from datetime import datetime
93 logger = Logger(verbose=not args.quiet)
95 # Setup signal handlers for graceful shutdown (ENH-183)
96 global _sprint_shutdown_requested
97 _sprint_shutdown_requested = False # Reset in case of multiple runs
98 signal.signal(signal.SIGINT, _sprint_signal_handler)
99 signal.signal(signal.SIGTERM, _sprint_signal_handler)
101 handoff_threshold = getattr(args, "handoff_threshold", None)
102 if handoff_threshold is not None:
103 os.environ["LL_HANDOFF_THRESHOLD"] = str(handoff_threshold)
105 context_limit = getattr(args, "context_limit", None)
106 if context_limit is not None:
107 os.environ["LL_CONTEXT_LIMIT"] = str(context_limit)
109 sprint = manager.load(args.sprint)
110 if not sprint:
111 logger.error(f"Sprint not found: {args.sprint}")
112 return 1
114 # Apply skip filter if provided
115 issues_to_process = list(sprint.issues)
116 skip_ids = parse_issue_ids(args.skip)
117 if skip_ids:
118 original_count = len(issues_to_process)
119 issues_to_process = [i for i in issues_to_process if i not in skip_ids]
120 skipped = original_count - len(issues_to_process)
121 if skipped > 0:
122 logger.info(f"Skipping {skipped} issue(s): {', '.join(sorted(skip_ids))}")
124 # Apply only filter if provided
125 only_ids = parse_issue_ids(getattr(args, "only", None))
126 if only_ids:
127 invalid_only = {p for p in only_ids if not any(_id_matches(i, p) for i in sprint.issues)}
128 if invalid_only:
129 logger.error(
130 f"Issue(s) not found in sprint definition: {', '.join(sorted(invalid_only))}"
131 )
132 return 1
133 issues_to_process = [
134 i for i in issues_to_process if any(_id_matches(i, p) for p in only_ids)
135 ]
136 logger.info(
137 f"Processing only {len(issues_to_process)} issue(s): {', '.join(sorted(only_ids))}"
138 )
140 # Apply type filter if provided
141 type_prefixes = parse_issue_types(getattr(args, "type", None))
142 if type_prefixes:
143 original_count = len(issues_to_process)
144 issues_to_process = [i for i in issues_to_process if i.split("-", 1)[0] in type_prefixes]
145 filtered = original_count - len(issues_to_process)
146 if filtered > 0:
147 logger.info(f"Filtered {filtered} issue(s) by type: {', '.join(sorted(type_prefixes))}")
149 # Pre-validate: skip issues already moved to completed/ (ENH-581)
150 pre_completed_skipped: list[str] = []
151 if config is not None:
152 completed_dir = config.get_completed_dir()
153 if completed_dir.exists():
154 still_active: list[str] = []
155 for issue_id in issues_to_process:
156 if list(completed_dir.glob(f"*-{issue_id}-*.md")):
157 logger.info(f" {issue_id}: already in completed/, skipping")
158 pre_completed_skipped.append(issue_id)
159 else:
160 still_active.append(issue_id)
161 if pre_completed_skipped:
162 logger.info(
163 f"Pre-validation: {len(pre_completed_skipped)} issue(s) already completed, "
164 f"{len(still_active)} active"
165 )
166 issues_to_process = still_active
168 if pre_completed_skipped and not issues_to_process:
169 logger.info("All sprint issues already completed - nothing to process")
170 return 0
172 # Validate issues exist
173 valid = manager.validate_issues(issues_to_process)
174 invalid = set(issues_to_process) - set(valid.keys())
176 if invalid:
177 logger.error(f"Issue IDs not found: {', '.join(sorted(invalid))}")
178 logger.info("Cannot execute sprint with missing issues")
179 return 1
181 # Load full IssueInfo objects for dependency analysis
182 issue_infos = manager.load_issue_infos(issues_to_process)
183 if not issue_infos:
184 logger.error("No issue files found")
185 return 1
187 # Gather all issue IDs on disk to avoid false "nonexistent" warnings
188 from little_loops.dependency_mapper import gather_all_issue_ids
190 issues_dir = config.project_root / config.issues.base_dir
191 all_known_ids = gather_all_issue_ids(issues_dir, config=config)
193 # Dependency analysis (ENH-301)
194 if not getattr(args, "skip_analysis", False):
195 from little_loops.dependency_mapper import analyze_dependencies
197 issue_contents = _build_issue_contents(issue_infos)
198 dep_config = config.dependency_mapping
199 dep_report = analyze_dependencies(
200 issue_infos, issue_contents, all_known_ids=all_known_ids, config=dep_config
201 )
202 _render_dependency_analysis(dep_report, logger, config=dep_config)
204 # Build dependency graph
205 dep_graph = DependencyGraph.from_issues(issue_infos, all_known_ids=all_known_ids)
207 # Detect cycles
208 if dep_graph.has_cycles():
209 cycles = dep_graph.detect_cycles()
210 for cycle in cycles:
211 logger.error(f"Dependency cycle detected: {' -> '.join(cycle)}")
212 return 1
214 # Get execution waves
215 try:
216 waves = dep_graph.get_execution_waves()
217 except ValueError as e:
218 logger.error(str(e))
219 return 1
221 # Refine waves for file overlap (ENH-306)
222 waves, contention_notes = refine_waves_for_contention(waves, config=config.dependency_mapping)
224 # Display execution plan
225 logger.info(f"Running sprint: {sprint.name}")
226 logger.info("Dependency analysis:")
227 for i, wave in enumerate(waves, 1):
228 issue_ids = ", ".join(issue.issue_id for issue in wave)
229 note = contention_notes[i - 1] if contention_notes else None
230 if note:
231 logger.info(
232 f" Wave {i}: {issue_ids}"
233 f" [sub-wave {note.sub_wave_index + 1}/{note.total_sub_waves}]"
234 )
235 else:
236 logger.info(f" Wave {i}: {issue_ids}")
238 if args.dry_run:
239 logger.info("\nDry run mode - no changes will be made")
240 return 0
242 # Initialize or load state
243 state: SprintState
244 start_wave = 1
246 if args.resume:
247 loaded_state = _load_sprint_state(logger)
248 if loaded_state and loaded_state.sprint_name == args.sprint:
249 state = loaded_state
250 # Find first incomplete wave by checking completed issues
251 completed_set = set(state.completed_issues)
252 for i, wave in enumerate(waves, 1):
253 wave_issue_ids = {issue.issue_id for issue in wave}
254 if not wave_issue_ids.issubset(completed_set):
255 start_wave = i
256 break
257 else:
258 # All waves completed
259 logger.info("Sprint already completed - nothing to resume")
260 _cleanup_sprint_state(logger)
261 return 0
262 logger.info(f"Resuming from wave {start_wave}/{len(waves)}")
263 logger.info(f" Previously completed: {len(state.completed_issues)} issues")
264 else:
265 if loaded_state:
266 logger.warning(
267 f"State file is for sprint '{loaded_state.sprint_name}', "
268 f"not '{args.sprint}' - starting fresh"
269 )
270 else:
271 logger.warning("No valid state found - starting fresh")
272 state = SprintState(
273 sprint_name=args.sprint,
274 started_at=datetime.now().isoformat(),
275 )
276 else:
277 # Fresh start - delete any old state
278 _cleanup_sprint_state(logger)
279 state = SprintState(
280 sprint_name=args.sprint,
281 started_at=datetime.now().isoformat(),
282 )
284 # Track exit status for error handling (ENH-185)
285 exit_code = 0
287 try:
288 # Determine max workers
289 max_workers = args.max_workers or (sprint.options.max_workers if sprint.options else 2)
291 # Execute wave by wave
292 completed: set[str] = set(state.completed_issues)
293 failed_waves = 0
294 total_duration = 0.0
295 total_waves = len(waves)
297 for wave_num, wave in enumerate(waves, 1):
298 # Check for shutdown request (ENH-183)
299 if _sprint_shutdown_requested:
300 logger.warning("Shutdown requested - saving state and exiting")
301 _save_sprint_state(state, logger)
302 exit_code = 1
303 return exit_code
305 # Skip already-completed waves when resuming
306 if wave_num < start_wave:
307 continue
309 wave_ids = [issue.issue_id for issue in wave]
310 state.current_wave = wave_num
311 logger.info(f"\nProcessing wave {wave_num}/{total_waves}: {', '.join(wave_ids)}")
313 wave_note = (
314 contention_notes[wave_num - 1]
315 if contention_notes and wave_num - 1 < len(contention_notes)
316 else None
317 )
318 is_contention_subwave = wave_note is not None
320 if len(wave) == 1 or is_contention_subwave:
321 # Single issue OR contention sub-wave — process in-place sequentially
322 # (contention sub-waves are displayed as "serialized steps" so must run that way)
323 from little_loops.issue_manager import process_issue_inplace
325 wave_failed = False
326 for issue in wave:
327 issue_result = process_issue_inplace(
328 info=issue,
329 config=config,
330 logger=logger,
331 dry_run=args.dry_run,
332 )
333 total_duration += issue_result.duration
334 if issue_result.success:
335 completed.add(issue.issue_id)
336 state.completed_issues.append(issue.issue_id)
337 state.timing[issue.issue_id] = {"total": issue_result.duration}
338 logger.success(f" {issue.issue_id}: completed")
339 elif issue_result.was_blocked:
340 completed.add(issue.issue_id)
341 state.skipped_blocked_issues[issue.issue_id] = issue_result.failure_reason
342 logger.warning(f" {issue.issue_id}: skipped (blocked by open dependency)")
343 else:
344 wave_failed = True
345 completed.add(issue.issue_id)
346 state.failed_issues[issue.issue_id] = "Issue processing failed"
347 logger.warning(f" {issue.issue_id}: failed")
348 if wave_failed:
349 failed_waves += 1
350 logger.warning(f"Wave {wave_num}/{total_waves} had failures")
351 else:
352 logger.success(
353 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}"
354 )
355 _save_sprint_state(state, logger)
356 if wave_num < total_waves:
357 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...")
358 # Check for shutdown before next wave (ENH-183)
359 if _sprint_shutdown_requested:
360 logger.warning("Shutdown requested - exiting after wave completion")
361 exit_code = 1
362 return exit_code
363 else:
364 # Multi-issue — use ParallelOrchestrator with worktrees
365 only_ids = set(wave_ids)
366 # Detect current branch for rebase/merge operations (BUG-439)
367 _br = subprocess.run(
368 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
369 capture_output=True,
370 text=True,
371 cwd=Path.cwd(),
372 )
373 _base_branch = _br.stdout.strip() if _br.returncode == 0 else "main"
374 # Runtime overlap detection disabled for sprints (ENH-512):
375 # refine_waves_for_contention() already splits overlapping
376 # issues into separate sub-waves before dispatch.
377 parallel_config = config.create_parallel_config(
378 max_workers=min(max_workers, len(wave)),
379 only_ids=only_ids,
380 dry_run=args.dry_run,
381 overlap_detection=False,
382 serialize_overlapping=True,
383 base_branch=_base_branch,
384 clean_start=True, # Sprint manages its own state; don't load stale orchestrator state
385 )
387 from little_loops.events import EventBus
388 from little_loops.extension import wire_extensions
390 event_bus = EventBus()
391 wire_extensions(event_bus, config.extensions)
392 orchestrator = ParallelOrchestrator(
393 parallel_config,
394 config,
395 Path.cwd(),
396 wave_label=f"Wave {wave_num}/{total_waves}",
397 event_bus=event_bus,
398 )
399 result = orchestrator.run()
400 total_duration += orchestrator.execution_duration
402 # Track completed/failed from this wave using per-issue results
403 actually_completed = set(orchestrator.queue.completed_ids)
404 actually_failed = set(orchestrator.queue.failed_ids)
406 for issue_id in wave_ids:
407 if issue_id in actually_completed:
408 completed.add(issue_id)
409 state.completed_issues.append(issue_id)
410 state.timing[issue_id] = {
411 "total": orchestrator.execution_duration / len(wave)
412 }
413 elif issue_id in actually_failed:
414 completed.add(issue_id)
415 state.failed_issues[issue_id] = "Issue failed during wave execution"
416 # else: issue was neither completed nor failed (interrupted/stranded)
417 # — leave untracked so it can be retried on resume
419 # Sequential retry for failed issues (ENH-308)
420 if actually_failed:
421 logger.info(f"Retrying {len(actually_failed)} failed issue(s) sequentially...")
422 from little_loops.issue_manager import process_issue_inplace
424 retried_ok = 0
425 for issue in wave:
426 if issue.issue_id not in actually_failed:
427 continue
428 logger.info(f" Retrying {issue.issue_id} in-place...")
429 retry_result = process_issue_inplace(
430 info=issue,
431 config=config,
432 logger=logger,
433 dry_run=args.dry_run,
434 )
435 total_duration += retry_result.duration
436 if retry_result.success:
437 retried_ok += 1
438 state.failed_issues.pop(issue.issue_id, None)
439 state.completed_issues.append(issue.issue_id)
440 state.timing[issue.issue_id] = {"total": retry_result.duration}
441 logger.success(f" Retry succeeded: {issue.issue_id}")
442 elif retry_result.was_blocked:
443 state.failed_issues.pop(issue.issue_id, None)
444 state.skipped_blocked_issues[issue.issue_id] = (
445 retry_result.failure_reason
446 )
447 logger.warning(
448 f" Retry skipped: {issue.issue_id} (blocked by open dependency)"
449 )
450 else:
451 logger.warning(f" Retry failed: {issue.issue_id}")
452 if retried_ok > 0:
453 logger.info(
454 f"Sequential retry recovered {retried_ok}/{len(actually_failed)} issue(s)"
455 )
457 # Check whether failures remain after retry (ENH-308)
458 remaining_failures = {iid for iid in actually_failed if iid in state.failed_issues}
459 if result == 0 or not remaining_failures:
460 logger.success(
461 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}"
462 )
463 else:
464 failed_waves += 1
465 logger.warning(f"Wave {wave_num}/{total_waves} had failures")
466 _save_sprint_state(state, logger)
467 if wave_num < total_waves:
468 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...")
469 # Check for shutdown before next wave (ENH-183)
470 if _sprint_shutdown_requested:
471 logger.warning("Shutdown requested - exiting after wave completion")
472 exit_code = 1
473 return exit_code
475 wave_word = "wave" if len(waves) == 1 else "waves"
476 skip_msg = (
477 f", {len(pre_completed_skipped)} already completed (skipped)"
478 if pre_completed_skipped
479 else ""
480 )
481 blocked_msg = (
482 f", {len(state.skipped_blocked_issues)} skipped (blocked)"
483 if state.skipped_blocked_issues
484 else ""
485 )
486 logger.info(
487 f"\nSprint completed: {len(completed)} issues processed "
488 f"({len(waves)} {wave_word}){skip_msg}{blocked_msg}"
489 )
490 logger.timing(f"Total execution time: {format_duration(total_duration)}")
491 if failed_waves > 0:
492 logger.warning(f"{failed_waves} wave(s) had failures")
493 exit_code = 1
494 else:
495 # Clean up state on successful completion
496 _cleanup_sprint_state(logger)
498 except KeyboardInterrupt:
499 # Belt-and-suspenders with signal handler (ENH-185)
500 logger.warning("Sprint interrupted by user (KeyboardInterrupt)")
501 exit_code = 130
503 except Exception as e:
504 # Catch unexpected exceptions (ENH-185)
505 logger.error(f"Sprint failed unexpectedly: {e}")
506 exit_code = 1
508 finally:
509 # Guaranteed state save on any non-success exit (ENH-185)
510 if exit_code != 0:
511 _save_sprint_state(state, logger)
512 logger.info("State saved before exit")
514 return exit_code