Coverage for little_loops / cli / sprint / run.py: 0%
324 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -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.output import use_color_enabled
14from little_loops.cli.sprint._helpers import _build_issue_contents, _render_dependency_analysis
15from little_loops.cli_args import _id_matches, parse_issue_ids, parse_issue_types, parse_labels
16from little_loops.dependency_graph import DependencyGraph, refine_waves_for_contention
17from little_loops.logger import Logger, format_duration
18from little_loops.parallel.orchestrator import ParallelOrchestrator
19from little_loops.sprint import SprintManager, SprintState
21if TYPE_CHECKING:
22 import argparse
24 from little_loops.config import BRConfig
26# Module-level shutdown flag for ll-sprint signal handling (ENH-183)
27_sprint_shutdown_requested: bool = False
28_sprint_logger: Logger | None = None
31def _sprint_signal_handler(signum: int, frame: FrameType | None) -> None:
32 """Handle shutdown signals gracefully for ll-sprint.
34 First signal: Set shutdown flag for graceful exit after current wave.
35 Second signal: Force immediate exit.
36 """
37 global _sprint_shutdown_requested
38 if _sprint_shutdown_requested:
39 # Second signal - force exit
40 msg = "\nForce shutdown requested"
41 if _sprint_logger is not None:
42 _sprint_logger.warning(msg)
43 else:
44 print(msg, file=sys.stderr)
45 sys.exit(1)
46 _sprint_shutdown_requested = True
47 msg = "\nShutdown requested, will exit after current wave..."
48 if _sprint_logger is not None:
49 _sprint_logger.warning(msg)
50 else:
51 print(msg, file=sys.stderr)
54def _get_sprint_state_file() -> Path:
55 """Get path to sprint state file."""
56 return Path.cwd() / ".sprint-state.json"
59def _load_sprint_state(logger: Logger) -> SprintState | None:
60 """Load sprint state from file."""
61 import json
63 state_file = _get_sprint_state_file()
64 if not state_file.exists():
65 return None
66 try:
67 data = json.loads(state_file.read_text())
68 state = SprintState.from_dict(data)
69 logger.info(f"State loaded from {state_file}")
70 return state
71 except (json.JSONDecodeError, KeyError) as e:
72 logger.warning(f"Failed to load state: {e}")
73 return None
76def _save_sprint_state(state: SprintState, logger: Logger) -> None:
77 """Save sprint state to file."""
78 import json
79 from datetime import datetime
81 state.last_checkpoint = datetime.now().isoformat()
82 state_file = _get_sprint_state_file()
83 state_file.write_text(json.dumps(state.to_dict(), indent=2))
84 logger.info(f"State saved to {state_file}")
87def _cleanup_sprint_state(logger: Logger) -> None:
88 """Remove sprint state file."""
89 state_file = _get_sprint_state_file()
90 if state_file.exists():
91 state_file.unlink()
92 logger.info("Sprint state file cleaned up")
95def _cmd_sprint_run(
96 args: argparse.Namespace,
97 manager: SprintManager,
98 config: BRConfig,
99) -> int:
100 """Execute a sprint with dependency-aware scheduling."""
101 from datetime import datetime
103 global _sprint_logger
104 logger = Logger(verbose=not args.quiet, use_color=use_color_enabled())
105 _sprint_logger = logger
107 # Setup signal handlers for graceful shutdown (ENH-183)
108 global _sprint_shutdown_requested
109 _sprint_shutdown_requested = False # Reset in case of multiple runs
110 signal.signal(signal.SIGINT, _sprint_signal_handler)
111 signal.signal(signal.SIGTERM, _sprint_signal_handler)
113 handoff_threshold = getattr(args, "handoff_threshold", None)
114 if handoff_threshold is not None:
115 os.environ["LL_HANDOFF_THRESHOLD"] = str(handoff_threshold)
117 context_limit = getattr(args, "context_limit", None)
118 if context_limit is not None:
119 os.environ["LL_CONTEXT_LIMIT"] = str(context_limit)
121 sprint = manager.load_or_resolve(args.sprint)
122 if not sprint:
123 logger.error(f"Sprint not found: {args.sprint}")
124 return 1
126 if not sprint.issues:
127 logger.info(f"Sprint '{sprint.name}' has no active issues — nothing to process")
128 return 0
130 if getattr(args, "save", False):
131 saved_path = sprint.save(manager.sprints_dir)
132 logger.info(f"Sprint saved to {saved_path}")
134 # Apply skip filter if provided
135 issues_to_process = list(sprint.issues)
136 skip_ids = parse_issue_ids(args.skip)
137 if skip_ids:
138 original_count = len(issues_to_process)
139 issues_to_process = [i for i in issues_to_process if i not in skip_ids]
140 skipped = original_count - len(issues_to_process)
141 if skipped > 0:
142 logger.info(f"Skipping {skipped} issue(s): {', '.join(sorted(skip_ids))}")
144 # Apply only filter if provided
145 only_ids = parse_issue_ids(getattr(args, "only", None))
146 if only_ids:
147 invalid_only = {p for p in only_ids if not any(_id_matches(i, p) for i in sprint.issues)}
148 if invalid_only:
149 logger.error(
150 f"Issue(s) not found in sprint definition: {', '.join(sorted(invalid_only))}"
151 )
152 return 1
153 issues_to_process = [
154 i for i in issues_to_process if any(_id_matches(i, p) for p in only_ids)
155 ]
156 logger.info(
157 f"Processing only {len(issues_to_process)} issue(s): {', '.join(sorted(only_ids))}"
158 )
160 # Apply type filter if provided
161 type_prefixes = parse_issue_types(getattr(args, "type", None))
162 if type_prefixes:
163 original_count = len(issues_to_process)
164 issues_to_process = [i for i in issues_to_process if i.split("-", 1)[0] in type_prefixes]
165 filtered = original_count - len(issues_to_process)
166 if filtered > 0:
167 logger.info(f"Filtered {filtered} issue(s) by type: {', '.join(sorted(type_prefixes))}")
169 # Pre-validate: skip issues with status: done/cancelled (ENH-1424)
170 pre_completed_skipped: list[str] = []
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 # Write milestone: field back to issue files for all valid issues in this sprint
182 from little_loops.frontmatter import parse_frontmatter, update_frontmatter
184 for _issue_id, path in valid.items():
185 content = path.read_text(encoding="utf-8")
186 updated = update_frontmatter(content, {"milestone": sprint.name})
187 if updated != content:
188 path.write_text(updated, encoding="utf-8")
189 logger.info(f"Set milestone: {sprint.name!r} on {len(valid)} issue file(s)")
191 # Skip issues already completed via frontmatter status (ENH-1424)
192 still_active: list[str] = []
193 for issue_id in issues_to_process:
194 path = valid[issue_id]
195 fm = parse_frontmatter(path.read_text(encoding="utf-8"))
196 if fm.get("status", "open") in ("done", "cancelled"):
197 logger.info(f" {issue_id}: already completed (status: {fm.get('status')}), skipping")
198 pre_completed_skipped.append(issue_id)
199 else:
200 still_active.append(issue_id)
201 if pre_completed_skipped:
202 logger.info(
203 f"Pre-validation: {len(pre_completed_skipped)} issue(s) already completed, "
204 f"{len(still_active)} active"
205 )
206 issues_to_process = still_active
208 if pre_completed_skipped and not issues_to_process:
209 logger.info("All sprint issues already completed - nothing to process")
210 return 0
212 # Load full IssueInfo objects for dependency analysis
213 issue_infos = manager.load_issue_infos(issues_to_process)
214 if not issue_infos:
215 logger.error("No issue files found")
216 return 1
218 # Apply label filter if provided (must run after IssueInfo load since labels come from frontmatter)
219 label_filter = parse_labels(getattr(args, "label", None))
220 if label_filter:
221 original_count = len(issue_infos)
222 issue_infos = [i for i in issue_infos if any(lb.lower() in label_filter for lb in i.labels)]
223 filtered = original_count - len(issue_infos)
224 if filtered > 0:
225 logger.info(f"Filtered {filtered} issue(s) by label: {', '.join(sorted(label_filter))}")
226 issues_to_process = [i.issue_id for i in issue_infos]
228 # Gather all issue IDs on disk to avoid false "nonexistent" warnings
229 from little_loops.dependency_mapper import gather_all_issue_ids
231 issues_dir = config.project_root / config.issues.base_dir
232 all_known_ids = gather_all_issue_ids(issues_dir, config=config)
234 # Dependency analysis (ENH-301)
235 if not getattr(args, "skip_analysis", False):
236 from little_loops.dependency_mapper import analyze_dependencies
238 issue_contents = _build_issue_contents(issue_infos)
239 dep_config = config.dependency_mapping
240 dep_report = analyze_dependencies(
241 issue_infos, issue_contents, all_known_ids=all_known_ids, config=dep_config
242 )
243 _render_dependency_analysis(dep_report, logger, config=dep_config)
245 # Build dependency graph
246 dep_graph = DependencyGraph.from_issues(issue_infos, all_known_ids=all_known_ids)
248 # Detect cycles
249 if dep_graph.has_cycles():
250 cycles = dep_graph.detect_cycles()
251 for cycle in cycles:
252 logger.error(f"Dependency cycle detected: {' -> '.join(cycle)}")
253 return 1
255 # Get execution waves
256 try:
257 waves = dep_graph.get_execution_waves()
258 except ValueError as e:
259 logger.error(str(e))
260 return 1
262 # Refine waves for file overlap (ENH-306)
263 waves, contention_notes = refine_waves_for_contention(waves, config=config.dependency_mapping)
265 # Display execution plan
266 logger.info(f"Running sprint: {sprint.name}")
267 logger.info("Dependency analysis:")
268 for i, wave in enumerate(waves, 1):
269 issue_ids = ", ".join(issue.issue_id for issue in wave)
270 note = contention_notes[i - 1] if contention_notes else None
271 if note:
272 logger.info(
273 f" Wave {i}: {issue_ids}"
274 f" [sub-wave {note.sub_wave_index + 1}/{note.total_sub_waves}]"
275 )
276 else:
277 logger.info(f" Wave {i}: {issue_ids}")
279 if args.dry_run:
280 logger.info("\nDry run mode - no changes will be made")
281 return 0
283 # Initialize or load state
284 state: SprintState
285 start_wave = 1
287 if args.resume:
288 loaded_state = _load_sprint_state(logger)
289 if loaded_state and loaded_state.sprint_name == sprint.name:
290 state = loaded_state
291 # Find first incomplete wave by checking completed issues
292 completed_set = set(state.completed_issues)
293 for i, wave in enumerate(waves, 1):
294 wave_issue_ids = {issue.issue_id for issue in wave}
295 if not wave_issue_ids.issubset(completed_set):
296 start_wave = i
297 break
298 else:
299 # All waves completed
300 logger.info("Sprint already completed - nothing to resume")
301 _cleanup_sprint_state(logger)
302 return 0
303 logger.info(f"Resuming from wave {start_wave}/{len(waves)}")
304 logger.info(f" Previously completed: {len(state.completed_issues)} issues")
305 else:
306 if loaded_state:
307 logger.warning(
308 f"State file is for sprint '{loaded_state.sprint_name}', "
309 f"not '{sprint.name}' - starting fresh"
310 )
311 else:
312 logger.warning("No valid state found - starting fresh")
313 state = SprintState(
314 sprint_name=sprint.name,
315 started_at=datetime.now().isoformat(),
316 )
317 else:
318 # Fresh start - delete any old state
319 _cleanup_sprint_state(logger)
320 state = SprintState(
321 sprint_name=sprint.name,
322 started_at=datetime.now().isoformat(),
323 )
325 # Track exit status for error handling (ENH-185)
326 exit_code = 0
328 try:
329 # Determine max workers
330 max_workers = args.max_workers or (sprint.options.max_workers if sprint.options else 2)
332 # Execute wave by wave
333 completed: set[str] = set(state.completed_issues)
334 failed_waves = 0
335 total_duration = 0.0
336 total_waves = len(waves)
338 for wave_num, wave in enumerate(waves, 1):
339 # Check for shutdown request (ENH-183)
340 if _sprint_shutdown_requested:
341 logger.warning("Shutdown requested - saving state and exiting")
342 _save_sprint_state(state, logger)
343 exit_code = 1
344 return exit_code
346 # Skip already-completed waves when resuming
347 if wave_num < start_wave:
348 continue
350 wave_ids = [issue.issue_id for issue in wave]
351 state.current_wave = wave_num
352 logger.info(f"\nProcessing wave {wave_num}/{total_waves}: {', '.join(wave_ids)}")
354 wave_note = (
355 contention_notes[wave_num - 1]
356 if contention_notes and wave_num - 1 < len(contention_notes)
357 else None
358 )
359 is_contention_subwave = wave_note is not None
361 if len(wave) == 1 or is_contention_subwave:
362 # Single issue OR contention sub-wave — process in-place sequentially
363 # (contention sub-waves are displayed as "serialized steps" so must run that way)
364 from little_loops.issue_manager import process_issue_inplace
366 wave_failed = False
367 for issue in wave:
368 # TODO(ENH-1686): sprint sequential path not yet live-written
369 issue_result = process_issue_inplace(
370 info=issue,
371 config=config,
372 logger=logger,
373 dry_run=args.dry_run,
374 )
375 total_duration += issue_result.duration
376 if issue_result.success:
377 completed.add(issue.issue_id)
378 state.completed_issues.append(issue.issue_id)
379 state.timing[issue.issue_id] = {"total": issue_result.duration}
380 logger.success(f" {issue.issue_id}: completed")
381 elif issue_result.was_blocked:
382 completed.add(issue.issue_id)
383 state.skipped_blocked_issues[issue.issue_id] = issue_result.failure_reason
384 logger.warning(f" {issue.issue_id}: skipped (blocked by open dependency)")
385 else:
386 wave_failed = True
387 completed.add(issue.issue_id)
388 state.failed_issues[issue.issue_id] = "Issue processing failed"
389 logger.warning(f" {issue.issue_id}: failed")
390 if wave_failed:
391 failed_waves += 1
392 logger.warning(f"Wave {wave_num}/{total_waves} had failures")
393 else:
394 logger.success(
395 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}"
396 )
397 _save_sprint_state(state, logger)
398 if wave_num < total_waves:
399 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...")
400 # Check for shutdown before next wave (ENH-183)
401 if _sprint_shutdown_requested:
402 logger.warning("Shutdown requested - exiting after wave completion")
403 exit_code = 1
404 return exit_code
405 else:
406 # Multi-issue — use ParallelOrchestrator with worktrees
407 only_ids = set(wave_ids)
408 # Detect current branch for rebase/merge operations (BUG-439)
409 _br = subprocess.run(
410 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
411 capture_output=True,
412 text=True,
413 cwd=Path.cwd(),
414 )
415 _base_branch = _br.stdout.strip() if _br.returncode == 0 else "main"
416 # Runtime overlap detection disabled for sprints (ENH-512):
417 # refine_waves_for_contention() already splits overlapping
418 # issues into separate sub-waves before dispatch.
419 parallel_config = config.create_parallel_config(
420 max_workers=min(max_workers, len(wave)),
421 only_ids=only_ids,
422 dry_run=args.dry_run,
423 overlap_detection=False,
424 serialize_overlapping=True,
425 base_branch=_base_branch,
426 clean_start=True, # Sprint manages its own state; don't load stale orchestrator state
427 )
429 from little_loops.events import EventBus
430 from little_loops.extension import wire_extensions
431 from little_loops.transport import wire_transports
433 event_bus = EventBus()
434 wire_extensions(event_bus, config.extensions)
435 wire_transports(event_bus, config.events)
436 orchestrator = ParallelOrchestrator(
437 parallel_config,
438 config,
439 Path.cwd(),
440 wave_label=f"Wave {wave_num}/{total_waves}",
441 event_bus=event_bus,
442 )
443 result = orchestrator.run()
444 total_duration += orchestrator.execution_duration
446 # Track completed/failed from this wave using per-issue results
447 actually_completed = set(orchestrator.queue.completed_ids)
448 actually_failed = set(orchestrator.queue.failed_ids)
450 for issue_id in wave_ids:
451 if issue_id in actually_completed:
452 completed.add(issue_id)
453 state.completed_issues.append(issue_id)
454 state.timing[issue_id] = {
455 "total": orchestrator.execution_duration / len(wave)
456 }
457 elif issue_id in actually_failed:
458 completed.add(issue_id)
459 state.failed_issues[issue_id] = "Issue failed during wave execution"
460 # else: issue was neither completed nor failed (interrupted/stranded)
461 # — leave untracked so it can be retried on resume
463 # Sequential retry for failed issues (ENH-308)
464 if actually_failed:
465 logger.info(f"Retrying {len(actually_failed)} failed issue(s) sequentially...")
466 from little_loops.issue_manager import process_issue_inplace
468 retried_ok = 0
469 for issue in wave:
470 if issue.issue_id not in actually_failed:
471 continue
472 logger.info(f" Retrying {issue.issue_id} in-place...")
473 # TODO(ENH-1686): sprint sequential path not yet live-written
474 retry_result = process_issue_inplace(
475 info=issue,
476 config=config,
477 logger=logger,
478 dry_run=args.dry_run,
479 )
480 total_duration += retry_result.duration
481 if retry_result.success:
482 retried_ok += 1
483 state.failed_issues.pop(issue.issue_id, None)
484 state.completed_issues.append(issue.issue_id)
485 state.timing[issue.issue_id] = {"total": retry_result.duration}
486 logger.success(f" Retry succeeded: {issue.issue_id}")
487 elif retry_result.was_blocked:
488 state.failed_issues.pop(issue.issue_id, None)
489 state.skipped_blocked_issues[issue.issue_id] = (
490 retry_result.failure_reason
491 )
492 logger.warning(
493 f" Retry skipped: {issue.issue_id} (blocked by open dependency)"
494 )
495 else:
496 logger.warning(f" Retry failed: {issue.issue_id}")
497 if retried_ok > 0:
498 logger.info(
499 f"Sequential retry recovered {retried_ok}/{len(actually_failed)} issue(s)"
500 )
502 # Check whether failures remain after retry (ENH-308)
503 remaining_failures = {iid for iid in actually_failed if iid in state.failed_issues}
504 if result == 0 or not remaining_failures:
505 logger.success(
506 f"Wave {wave_num}/{total_waves} completed: {', '.join(wave_ids)}"
507 )
508 else:
509 failed_waves += 1
510 logger.warning(f"Wave {wave_num}/{total_waves} had failures")
511 _save_sprint_state(state, logger)
512 if wave_num < total_waves:
513 logger.info(f"Continuing to wave {wave_num + 1}/{total_waves}...")
514 # Check for shutdown before next wave (ENH-183)
515 if _sprint_shutdown_requested:
516 logger.warning("Shutdown requested - exiting after wave completion")
517 exit_code = 1
518 return exit_code
520 wave_word = "wave" if len(waves) == 1 else "waves"
521 skip_msg = (
522 f", {len(pre_completed_skipped)} already completed (skipped)"
523 if pre_completed_skipped
524 else ""
525 )
526 blocked_msg = (
527 f", {len(state.skipped_blocked_issues)} skipped (blocked)"
528 if state.skipped_blocked_issues
529 else ""
530 )
531 logger.info(
532 f"\nSprint completed: {len(completed)} issues processed "
533 f"({len(waves)} {wave_word}){skip_msg}{blocked_msg}"
534 )
535 logger.timing(f"Total execution time: {format_duration(total_duration)}")
536 if failed_waves > 0:
537 logger.warning(f"{failed_waves} wave(s) had failures")
538 exit_code = 1
539 else:
540 # Clean up state on successful completion
541 _cleanup_sprint_state(logger)
543 except KeyboardInterrupt:
544 # Belt-and-suspenders with signal handler (ENH-185)
545 logger.warning("Sprint interrupted by user (KeyboardInterrupt)")
546 exit_code = 130
548 except Exception as e:
549 # Catch unexpected exceptions (ENH-185)
550 logger.error(f"Sprint failed unexpectedly: {e}")
551 exit_code = 1
553 finally:
554 # Guaranteed state save on any non-success exit (ENH-185)
555 if exit_code != 0:
556 _save_sprint_state(state, logger)
557 logger.info("State saved before exit")
559 _sprint_logger = None # Clear for test isolation
560 return exit_code