Coverage for little_loops / parallel / orchestrator.py: 8%
674 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:08 -0500
1"""Main orchestrator for parallel issue processing.
3Coordinates the priority queue, worker pool, and merge coordinator to process
4multiple issues concurrently.
5"""
7from __future__ import annotations
9import json
10import os
11import re
12import shutil
13import signal
14import subprocess
15import tempfile
16import threading
17import time
18from datetime import UTC, datetime
19from pathlib import Path
20from typing import TYPE_CHECKING, Any
22from little_loops.events import EventBus
23from little_loops.frontmatter import update_frontmatter
24from little_loops.issue_parser import IssueInfo
25from little_loops.logger import Logger, format_duration
26from little_loops.parallel.git_lock import GitLock
27from little_loops.parallel.merge_coordinator import MergeCoordinator
28from little_loops.parallel.overlap_detector import OverlapDetector
29from little_loops.parallel.priority_queue import IssuePriorityQueue
30from little_loops.parallel.types import (
31 OrchestratorState,
32 ParallelConfig,
33 PendingWorktreeInfo,
34 WorkerResult,
35)
36from little_loops.parallel.worker_pool import WorkerPool
37from little_loops.session_log import append_session_log_entry
38from little_loops.worktree_utils import _is_ll_worktree
40if TYPE_CHECKING:
41 from little_loops.config import BRConfig
44class ParallelOrchestrator:
45 """Main controller for parallel issue processing.
47 Coordinates:
48 - Issue scanning and prioritization
49 - Worker dispatch (P0 sequential, P1-P5 parallel)
50 - Merge coordination
51 - State persistence for resume capability
52 - Graceful shutdown on signals
54 Example:
55 >>> from little_loops.config import BRConfig
56 >>> from little_loops.parallel import ParallelConfig, ParallelOrchestrator
57 >>> br_config = BRConfig(Path.cwd())
58 >>> parallel_config = ParallelConfig(max_workers=2)
59 >>> orchestrator = ParallelOrchestrator(parallel_config, br_config)
60 >>> exit_code = orchestrator.run()
61 """
63 def __init__(
64 self,
65 parallel_config: ParallelConfig,
66 br_config: BRConfig,
67 repo_path: Path | None = None,
68 verbose: bool = True,
69 wave_label: str | None = None,
70 event_bus: EventBus | None = None,
71 ) -> None:
72 """Initialize the orchestrator.
74 Args:
75 parallel_config: Parallel processing configuration
76 br_config: Project configuration
77 repo_path: Path to the git repository (default: current directory)
78 verbose: Whether to output progress messages
79 wave_label: Optional label for wave-based execution (e.g., "Wave 1")
80 event_bus: Optional EventBus for emitting worker completion events
81 """
82 self.parallel_config = parallel_config
83 self.br_config = br_config
84 self.repo_path = repo_path or Path.cwd()
85 from little_loops.cli.output import use_color_enabled
87 self.logger = Logger(verbose=verbose, use_color=use_color_enabled())
88 self.wave_label = wave_label
89 self._event_bus = event_bus
90 self._execution_duration: float = 0.0
92 # Create shared git lock for serializing main repo operations
93 # This prevents index.lock race conditions between workers and merge coordinator
94 self._git_lock = GitLock(self.logger)
96 # Initialize components with shared git lock
97 self.queue = IssuePriorityQueue()
98 self.worker_pool = WorkerPool(
99 parallel_config, br_config, self.logger, self.repo_path, self._git_lock
100 )
101 self.merge_coordinator = MergeCoordinator(
102 parallel_config, self.logger, self.repo_path, self._git_lock
103 )
105 # State management
106 self.state = OrchestratorState()
107 self._state_lock = threading.Lock()
108 self._shutdown_requested = False
109 self._original_sigint: Any = None
110 self._original_sigterm: Any = None
112 # Track issue info for lifecycle completion after merge
113 self._issue_info_by_id: dict[str, IssueInfo] = {}
114 # Track interrupted issues separately from failures (ENH-036)
115 self._interrupted_issues: list[str] = []
116 # Accumulate per-issue failure reasons for state file (BUG-1383)
117 self._worker_errors: dict[str, str] = {}
118 # Track feature-branch state when use_feature_branches=True (ENH-665, BUG-2172)
119 self._pr_ready_branches: dict[str, dict] = {} # issue_id -> {branch_name, pushed, pr_url}
121 # Overlap detection (ENH-143)
122 self.overlap_detector: OverlapDetector | None = (
123 OverlapDetector(config=br_config.dependency_mapping)
124 if parallel_config.overlap_detection
125 else None
126 )
127 # Track deferred issues for re-check after active issues complete
128 self._deferred_issues: list[IssueInfo] = []
129 # Track last status report time for progress visibility (ENH-262)
130 self._last_status_time: float = 0.0
131 self._last_status_line: str = ""
132 # Track last state save time to throttle disk writes (ENH-485)
133 self._last_save_time: float = 0.0
135 @property
136 def execution_duration(self) -> float:
137 """Return the total execution duration in seconds."""
138 return self._execution_duration
140 def run(self) -> int:
141 """Run the parallel issue processor.
143 Returns:
144 Exit code (0 = success, 1 = failure)
145 """
146 try:
147 self._setup_signal_handlers()
148 self._ensure_gitignore_entries()
150 # Check for pending work from previous runs (unless clean start)
151 if not self.parallel_config.clean_start:
152 pending_worktrees = self._check_pending_worktrees()
154 # Handle pending worktrees based on flags
155 pending_with_work = [p for p in pending_worktrees if p.has_pending_work]
156 if pending_with_work:
157 if self.parallel_config.merge_pending:
158 self._merge_pending_worktrees(pending_worktrees)
159 elif not self.parallel_config.ignore_pending:
160 # Default behavior: just report (cleanup happens below)
161 self.logger.info(
162 "Continuing with cleanup (use --merge-pending to merge)..."
163 )
165 self._cleanup_orphaned_worktrees()
166 self._load_state()
168 if self.parallel_config.dry_run:
169 return self._dry_run()
171 return self._execute()
173 except KeyboardInterrupt:
174 self.logger.warning("Interrupted by user")
175 return 1
176 except Exception as e:
177 self.logger.error(f"Fatal error: {e}")
178 return 1
179 finally:
180 self._cleanup()
181 self._restore_signal_handlers()
183 def _setup_signal_handlers(self) -> None:
184 """Setup signal handlers for graceful shutdown."""
185 self._original_sigint = signal.signal(signal.SIGINT, self._signal_handler)
186 self._original_sigterm = signal.signal(signal.SIGTERM, self._signal_handler)
188 def _restore_signal_handlers(self) -> None:
189 """Restore original signal handlers."""
190 if self._original_sigint is not None:
191 signal.signal(signal.SIGINT, self._original_sigint)
192 if self._original_sigterm is not None:
193 signal.signal(signal.SIGTERM, self._original_sigterm)
195 def _ensure_gitignore_entries(self) -> None:
196 """Ensure .gitignore has entries for parallel processing artifacts.
198 Adds entries for:
199 - .parallel-manage-state.json (state file)
200 - .worktrees/ (git worktree directory)
202 This prevents these files from being tracked by git, which would cause
203 conflicts during merge operations (state file is continuously updated).
204 """
205 gitignore_path = self.repo_path / ".gitignore"
206 required_entries = [
207 ".parallel-manage-state.json",
208 ".worktrees/",
209 ]
211 existing_content = ""
212 if gitignore_path.exists():
213 existing_content = gitignore_path.read_text()
215 # Check which entries are missing
216 missing_entries = []
217 for entry in required_entries:
218 # Check for exact match or pattern that would cover it
219 if entry not in existing_content:
220 missing_entries.append(entry)
222 if not missing_entries:
223 return
225 # Append missing entries
226 addition = "\n# ll-parallel artifacts\n"
227 for entry in missing_entries:
228 addition += f"{entry}\n"
230 # Ensure file ends with newline before adding
231 if existing_content and not existing_content.endswith("\n"):
232 addition = "\n" + addition
234 gitignore_path.write_text(existing_content + addition)
235 self.logger.info(f"Added {len(missing_entries)} entries to .gitignore")
237 def _cleanup_orphaned_worktrees(self) -> None:
238 """Clean up worktrees from previous interrupted runs.
240 Scans the worktree base directory and removes any worktrees that are
241 not from the current session. This handles cases where a previous run
242 was interrupted (Ctrl+C) and worktrees were not cleaned up.
243 """
244 worktree_base = self.repo_path / self.parallel_config.worktree_base
245 if not worktree_base.exists():
246 return
248 # Get list of worktree directories, skipping those owned by live processes (BUG-579)
249 orphaned = []
250 for item in worktree_base.iterdir():
251 if item.is_dir() and _is_ll_worktree(item.name):
252 # Check for a .ll-session-<pid> marker left by an active orchestrator
253 owned_by_live = False
254 for marker in item.glob(".ll-session-*"):
255 try:
256 pid = int(marker.name.split("-")[-1])
257 os.kill(pid, 0) # Signal 0: check if process exists
258 owned_by_live = True
259 break
260 except (ProcessLookupError, ValueError):
261 pass
262 except PermissionError:
263 owned_by_live = True # Process exists, no permission to signal
264 break
265 if owned_by_live:
266 self.logger.info(f"Skipping {item.name}: owned by running process")
267 continue
268 orphaned.append(item)
270 if orphaned:
271 self.logger.info(f"Cleaning up {len(orphaned)} orphaned worktree(s) from previous run")
273 for worktree_path in orphaned:
274 try:
275 self._git_lock.run(
276 ["worktree", "unlock", str(worktree_path)],
277 cwd=self.repo_path,
278 timeout=10,
279 )
280 # Try git worktree remove first
281 self._git_lock.run(
282 ["worktree", "remove", "--force", str(worktree_path)],
283 cwd=self.repo_path,
284 timeout=30,
285 )
287 # If git worktree remove failed, force delete the directory
288 if worktree_path.exists():
289 shutil.rmtree(worktree_path, ignore_errors=True)
291 # Try to delete the associated branch using the actual branch name
292 branch_result = subprocess.run(
293 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
294 cwd=worktree_path,
295 capture_output=True,
296 text=True,
297 )
298 branch_name = (
299 branch_result.stdout.strip() if branch_result.returncode == 0 else None
300 )
301 if branch_name and branch_name.startswith("parallel/"):
302 self._git_lock.run(
303 ["branch", "-D", branch_name],
304 cwd=self.repo_path,
305 timeout=10,
306 )
307 except Exception as e:
308 self.logger.warning(f"Failed to clean up {worktree_path.name}: {e}")
310 # Also prune git worktree references
311 self._git_lock.run(
312 ["worktree", "prune"],
313 cwd=self.repo_path,
314 timeout=30,
315 )
317 self._prune_ghost_worktree_refs()
319 def _prune_ghost_worktree_refs(self) -> None:
320 """Prune git worktree metadata entries whose on-disk path no longer exists.
322 Handles the SIGKILL race where a worker directory was deleted before
323 git worktree prune ran, leaving .git/worktrees/<name>/ intact. The
324 next git worktree add for the same path would then fail with "already exists".
325 """
326 try:
327 result = self._git_lock.run(
328 ["worktree", "list", "--porcelain"],
329 cwd=self.repo_path,
330 timeout=30,
331 )
332 except Exception as e:
333 self.logger.warning(f"Failed to list worktrees for ghost ref scan: {e}")
334 return
336 ghost_names: list[str] = []
337 current: dict[str, str] = {}
338 for line in result.stdout.splitlines():
339 if not line:
340 if current:
341 path_str = current.get("worktree", "")
342 name = Path(path_str).name
343 if _is_ll_worktree(name) and path_str and not Path(path_str).exists():
344 ghost_names.append(name)
345 current = {}
346 continue
347 key, _, value = line.partition(" ")
348 current[key] = value
349 if current:
350 path_str = current.get("worktree", "")
351 name = Path(path_str).name
352 if _is_ll_worktree(name) and path_str and not Path(path_str).exists():
353 ghost_names.append(name)
355 if not ghost_names:
356 return
358 for name in ghost_names:
359 self.logger.info(f"Pruned ghost ref: {name}")
361 try:
362 self._git_lock.run(
363 ["worktree", "prune"],
364 cwd=self.repo_path,
365 timeout=30,
366 )
367 except Exception as e:
368 self.logger.warning(f"Failed to prune ghost worktree refs: {e}")
370 def _inspect_worktree(self, worktree_path: Path) -> PendingWorktreeInfo | None:
371 """Inspect a worktree to determine its status.
373 Args:
374 worktree_path: Path to the worktree directory
376 Returns:
377 PendingWorktreeInfo if inspection succeeded, None if failed
378 """
379 try:
380 # Read actual branch name from worktree via rev-parse
381 branch_result = subprocess.run(
382 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
383 cwd=worktree_path,
384 capture_output=True,
385 text=True,
386 )
387 if branch_result.returncode == 0:
388 branch_name = branch_result.stdout.strip()
389 else:
390 # Fall back to string derivation if rev-parse fails
391 branch_name = worktree_path.name.replace("worker-", "parallel/")
393 # Extract issue ID (e.g., bug-045 -> BUG-045)
394 # Pattern: worker-<issue-id>-<timestamp>
395 match = re.match(r"worker-([a-z]+-\d+)-\d{8}-\d{6}", worktree_path.name)
396 issue_id = match.group(1).upper() if match else worktree_path.name
398 # Check commits ahead of main
399 result = self._git_lock.run(
400 ["rev-list", "--count", f"{self.parallel_config.base_branch}..{branch_name}"],
401 cwd=self.repo_path,
402 timeout=10,
403 )
404 commits_ahead = int(result.stdout.strip()) if result.returncode == 0 else 0
406 # Check for uncommitted changes in worktree
407 result = self._git_lock.run(
408 ["status", "--porcelain"],
409 cwd=worktree_path,
410 timeout=10,
411 )
412 changed_files = []
413 has_uncommitted = False
414 if result.returncode == 0 and result.stdout.strip():
415 has_uncommitted = True
416 changed_files = [line[3:] for line in result.stdout.strip().split("\n") if line]
418 return PendingWorktreeInfo(
419 worktree_path=worktree_path,
420 branch_name=branch_name,
421 issue_id=issue_id,
422 commits_ahead=commits_ahead,
423 has_uncommitted_changes=has_uncommitted,
424 changed_files=changed_files,
425 )
426 except Exception as e:
427 self.logger.warning(f"Failed to inspect worktree {worktree_path.name}: {e}")
428 return None
430 def _check_pending_worktrees(self) -> list[PendingWorktreeInfo]:
431 """Check for pending worktrees from previous runs and report status.
433 Returns:
434 List of pending worktree information
435 """
436 worktree_base = self.repo_path / self.parallel_config.worktree_base
437 if not worktree_base.exists():
438 return []
440 # Find all worker directories
441 worktrees = [
442 item for item in worktree_base.iterdir() if item.is_dir() and _is_ll_worktree(item.name)
443 ]
445 if not worktrees:
446 return []
448 self.logger.info("Checking for pending work from previous runs...")
450 # Inspect each worktree
451 pending_info: list[PendingWorktreeInfo] = []
452 for worktree_path in worktrees:
453 info = self._inspect_worktree(worktree_path)
454 if info:
455 pending_info.append(info)
457 # Report findings
458 with_work = [p for p in pending_info if p.has_pending_work]
459 if with_work:
460 self.logger.warning(f"Found {len(with_work)} worktree(s) with pending work:")
461 for info in with_work:
462 status_parts = []
463 if info.commits_ahead > 0:
464 status_parts.append(f"{info.commits_ahead} commit(s) ahead of main")
465 if info.has_uncommitted_changes:
466 status_parts.append(f"{len(info.changed_files)} uncommitted file(s)")
467 status = ", ".join(status_parts)
468 self.logger.warning(f" - {info.worktree_path.name}: {info.issue_id} ({status})")
470 self.logger.info("")
471 self.logger.info("Options:")
472 self.logger.info(" --merge-pending Attempt to merge pending work before continuing")
473 self.logger.info(" --clean-start Remove all worktrees and start fresh")
474 self.logger.info(
475 " --ignore-pending Continue without action (worktrees will be cleaned up)"
476 )
477 elif pending_info:
478 self.logger.info(f"Found {len(pending_info)} orphaned worktree(s) with no pending work")
480 return pending_info
482 def _merge_pending_worktrees(self, pending: list[PendingWorktreeInfo]) -> None:
483 """Attempt to merge pending worktrees from previous runs.
485 Args:
486 pending: List of pending worktree information
487 """
488 with_work = [p for p in pending if p.has_pending_work]
489 if not with_work:
490 return
492 self.logger.info(f"Attempting to merge {len(with_work)} pending worktree(s)...")
494 for info in with_work:
495 try:
496 # If there are uncommitted changes, commit them first
497 if info.has_uncommitted_changes:
498 self.logger.info(f" Committing uncommitted changes in {info.issue_id}...")
499 self._git_lock.run(
500 ["add", "-A"],
501 cwd=info.worktree_path,
502 timeout=30,
503 )
504 self._git_lock.run(
505 [
506 "commit",
507 "-m",
508 f"WIP: Auto-commit from interrupted session for {info.issue_id}",
509 ],
510 cwd=info.worktree_path,
511 timeout=30,
512 )
514 # Attempt merge
515 self.logger.info(f" Merging {info.issue_id} ({info.branch_name})...")
516 result = self._git_lock.run(
517 [
518 "merge",
519 "--no-ff",
520 info.branch_name,
521 "-m",
522 f"Merge pending work for {info.issue_id}",
523 ],
524 cwd=self.repo_path,
525 timeout=60,
526 )
528 if result.returncode == 0:
529 self.logger.success(f" Successfully merged {info.issue_id}")
530 # Clean up the worktree after successful merge
531 self._git_lock.run(
532 ["worktree", "remove", "--force", str(info.worktree_path)],
533 cwd=self.repo_path,
534 timeout=30,
535 )
536 self._git_lock.run(
537 ["branch", "-D", info.branch_name],
538 cwd=self.repo_path,
539 timeout=10,
540 )
541 else:
542 self.logger.warning(f" Failed to merge {info.issue_id}: {result.stderr}")
543 # Abort the merge if it failed
544 self._git_lock.run(
545 ["merge", "--abort"],
546 cwd=self.repo_path,
547 timeout=10,
548 )
550 except Exception as e:
551 self.logger.warning(f" Error merging {info.issue_id}: {e}")
553 def _signal_handler(self, signum: int, frame: object) -> None:
554 """Handle shutdown signals gracefully."""
555 self._shutdown_requested = True
556 # Propagate to worker pool for interrupted worker detection (ENH-036)
557 self.worker_pool.set_shutdown_requested(True)
558 self.logger.warning(f"Received signal {signum}, shutting down gracefully...")
560 def _load_state(self) -> None:
561 """Load state from file for resume capability."""
562 if self.parallel_config.clean_start:
563 self.state.started_at = datetime.now().isoformat()
564 return
565 state_file = self.repo_path / self.parallel_config.state_file
566 if not state_file.exists():
567 self.state.started_at = datetime.now().isoformat()
568 return
570 try:
571 data = json.loads(state_file.read_text())
572 self.state = OrchestratorState.from_dict(data)
574 # Restore queue state
575 self.queue.load_completed(self.state.completed_issues)
576 self.queue.load_failed(self.state.failed_issues.keys())
578 self.logger.info(
579 f"Resumed from previous state: "
580 f"{len(self.state.completed_issues)} completed, "
581 f"{len(self.state.failed_issues)} failed"
582 )
583 except Exception as e:
584 self.logger.warning(f"Could not load state: {e}")
585 self.state.started_at = datetime.now().isoformat()
587 def _save_state(self, force: bool = False) -> None:
588 """Save current state to file using an atomic write.
590 Writes are throttled to at most once every 5 seconds to reduce filesystem I/O
591 during high-frequency loop ticks (e.g., merge-waiting phase). Pass force=True
592 to bypass the throttle, e.g., on shutdown.
593 """
594 now = time.time()
595 if not force and now - self._last_save_time < 5.0:
596 return
597 self._last_save_time = now
598 with self._state_lock:
599 self.state.last_checkpoint = datetime.now().isoformat()
600 self.state.completed_issues = self.queue.completed_ids
601 self.state.failed_issues = {
602 issue_id: self._worker_errors.get(issue_id, "Failed")
603 for issue_id in self.queue.failed_ids
604 }
605 self.state.in_progress_issues = self.queue.in_progress_ids
607 state_file = self.repo_path / self.parallel_config.state_file
608 data = json.dumps(self.state.to_dict(), indent=2)
609 tmp_fd, tmp_path = tempfile.mkstemp(dir=state_file.parent, suffix=".tmp")
610 try:
611 with os.fdopen(tmp_fd, "w") as f:
612 f.write(data)
613 os.replace(tmp_path, state_file)
614 except Exception:
615 os.unlink(tmp_path)
616 raise
618 def _cleanup_state(self) -> None:
619 """Remove state file on successful completion."""
620 state_file = self.repo_path / self.parallel_config.state_file
621 if state_file.exists():
622 state_file.unlink()
624 def _dry_run(self) -> int:
625 """Preview what would be processed without executing.
627 Returns:
628 Exit code (always 0 for dry run)
629 """
630 issues = self._scan_issues()
632 self.logger.info("=" * 60)
633 self.logger.info("DRY RUN - No changes will be made")
634 self.logger.info("=" * 60)
635 self.logger.info("")
637 if not issues:
638 self.logger.info("No issues found matching criteria")
639 return 0
641 self.logger.info(f"Found {len(issues)} issues to process:")
642 self.logger.info("")
644 # Group by priority
645 by_priority: dict[str, list[IssueInfo]] = {}
646 for issue in issues:
647 by_priority.setdefault(issue.priority, []).append(issue)
649 for priority in IssuePriorityQueue.DEFAULT_PRIORITIES:
650 if priority not in by_priority:
651 continue
653 priority_issues = by_priority[priority]
654 self.logger.info(f" {priority} ({len(priority_issues)} issues):")
655 for issue in priority_issues:
656 mode = (
657 "sequential"
658 if priority == "P0" and self.parallel_config.p0_sequential
659 else "parallel"
660 )
661 self.logger.info(f" - {issue.issue_id}: {issue.title} [{mode}]")
663 self.logger.info("")
664 self.logger.info("Configuration:")
665 self.logger.info(f" Workers: {self.parallel_config.max_workers}")
666 self.logger.info(f" P0 Sequential: {self.parallel_config.p0_sequential}")
667 self.logger.info(f" Max Issues: {self.parallel_config.max_issues or 'unlimited'}")
668 self.logger.info(f" Command Prefix: {self.parallel_config.command_prefix}")
670 return 0
672 def _maybe_report_status(self) -> None:
673 """Report status if enough time has elapsed since last report.
675 Reports every 5 seconds during active processing for progress visibility (ENH-262).
676 Suppresses duplicate lines when nothing has changed.
677 """
678 now = time.time()
679 # Report every 5 seconds
680 if now - self._last_status_time < 5.0:
681 return
683 self._last_status_time = now
685 # Build status line
686 parts = []
688 # Add wave label if present
689 if self.wave_label:
690 parts.append(f"{self.wave_label}")
692 # Get queue counts
693 in_progress = len(self.queue.in_progress_ids)
694 completed = self.queue.completed_count
695 failed = self.queue.failed_count
696 pending_merge = self.merge_coordinator.pending_count
698 parts.append(f"Active: {in_progress}")
699 parts.append(f"Done: {completed}")
700 if failed > 0:
701 parts.append(f"Failed: {failed}")
702 if pending_merge > 0:
703 parts.append(f"Merging: {pending_merge}")
705 # Build status line
706 status = " | ".join(parts)
708 # Get active worker stages
709 active_stages = self.worker_pool.get_active_stages()
711 # Add worker details if any are active
712 if active_stages:
713 # Group by stage
714 by_stage: dict[str, list[str]] = {}
715 for issue_id, worker_stage in active_stages.items():
716 stage_name = worker_stage.value.title()
717 by_stage.setdefault(stage_name, []).append(issue_id)
719 stage_parts = []
720 for stage_name in ["Validating", "Implementing", "Verifying", "Merging"]:
721 if stage_name in by_stage:
722 issue_ids = ", ".join(by_stage[stage_name])
723 stage_parts.append(f"{stage_name}: [{issue_ids}]")
725 if stage_parts:
726 status += " | " + " | ".join(stage_parts)
728 # Skip if nothing changed since last report
729 if status == self._last_status_line:
730 return
731 self._last_status_line = status
733 # Log with gray color to distinguish from normal logs
734 self.logger.debug(status)
736 def _execute(self) -> int:
737 """Execute parallel issue processing.
739 Returns:
740 Exit code (0 = success, 1 = failure)
741 """
742 start_time = time.time()
744 # Scan and queue issues
745 issues = self._scan_issues()
746 if not issues:
747 self.logger.info("No issues to process")
748 return 0
750 # Store issue info for lifecycle completion after merge
751 for issue in issues:
752 self._issue_info_by_id[issue.issue_id] = issue
754 added = self.queue.add_many(issues)
755 self.logger.info(f"Queued {added} issues for processing")
757 # Start components
758 self.worker_pool.start()
759 self.merge_coordinator.start()
761 # Process issues
762 issues_processed = 0
763 max_issues = self.parallel_config.max_issues or float("inf")
765 while not self._shutdown_requested:
766 # Check if done
767 if self.queue.empty() and self.worker_pool.active_count == 0:
768 # Wait for pending merges
769 if self.merge_coordinator.pending_count == 0:
770 break
772 # Check max issues limit
773 if issues_processed >= max_issues:
774 self.logger.info(f"Reached max issues limit ({max_issues})")
775 break
777 # Get next issue if workers available
778 if self.worker_pool.active_count < self.parallel_config.max_workers:
779 queued = self.queue.get(block=False)
780 if queued:
781 issue = queued.issue_info
783 # P0 sequential processing
784 if issue.priority == "P0" and self.parallel_config.p0_sequential:
785 self._process_sequential(issue)
786 else:
787 self._process_parallel(issue)
789 issues_processed += 1
791 # Save state periodically
792 self._save_state()
794 # Report status periodically for progress visibility (ENH-262)
795 self._maybe_report_status()
797 # Small sleep to prevent busy loop
798 time.sleep(0.1)
800 # Wait for completion
801 self._wait_for_completion()
803 # Report results
804 self._report_results(start_time)
806 # Cleanup state on success
807 if not self._shutdown_requested and self.queue.failed_count == 0:
808 self._cleanup_state()
810 return 0 if self.queue.failed_count == 0 else 1
812 def _scan_issues(self) -> list[IssueInfo]:
813 """Scan for issues matching criteria.
815 Returns:
816 List of issues sorted by priority
817 """
818 # Combine skip_ids from state and config
819 skip_ids = set(self.state.completed_issues) | set(self.state.failed_issues.keys())
820 if self.parallel_config.skip_ids:
821 skip_ids |= self.parallel_config.skip_ids
823 issues = IssuePriorityQueue.scan_issues(
824 self.br_config,
825 priority_filter=list(self.parallel_config.priority_filter),
826 skip_ids=skip_ids,
827 only_ids=self.parallel_config.only_ids,
828 type_prefixes=self.parallel_config.type_prefixes,
829 label_filter=self.parallel_config.label_filter,
830 )
832 # Apply max issues limit
833 if self.parallel_config.max_issues > 0:
834 issues = issues[: self.parallel_config.max_issues]
836 return issues
838 def _process_sequential(self, issue: IssueInfo) -> None:
839 """Process an issue sequentially (blocking).
841 Args:
842 issue: Issue to process
843 """
844 self.logger.info(f"Processing {issue.issue_id} sequentially (P0)")
846 # Wait for any parallel work to finish
847 while self.worker_pool.active_count > 0:
848 time.sleep(0.5)
850 # Process in main repo (no worktree isolation needed)
851 # Note: No callback here - _merge_sequential handles the result explicitly
852 # to avoid double-processing (callback would also queue merge/close)
853 future = self.worker_pool.submit(issue)
855 # Wait for completion
856 try:
857 result = future.result(timeout=self.parallel_config.timeout_per_issue)
858 if result.success:
859 # Merge immediately for P0
860 self._merge_sequential(result)
861 except Exception as e:
862 self.logger.error(f"Sequential processing failed: {e}")
863 self.queue.mark_failed(issue.issue_id)
865 def _process_parallel(self, issue: IssueInfo) -> None:
866 """Process an issue in parallel (non-blocking).
868 Args:
869 issue: Issue to process
870 """
871 # Check for overlaps if enabled (ENH-143)
872 if self.overlap_detector:
873 overlap = self.overlap_detector.check_overlap(issue)
874 if overlap:
875 if self.parallel_config.serialize_overlapping:
876 self.logger.warning(
877 f"Deferring {issue.issue_id} - overlaps with {overlap.overlapping_issues}"
878 )
879 # Track for re-check when active issues complete
880 self._deferred_issues.append(issue)
881 return
882 else:
883 self.logger.warning(
884 f"Warning: {issue.issue_id} may conflict with {overlap.overlapping_issues}"
885 )
887 # Register as active before dispatch
888 self.overlap_detector.register_issue(issue)
890 self.logger.info(f"Dispatching {issue.issue_id} to worker pool")
891 self.worker_pool.submit(issue, self._on_worker_complete)
893 def _on_worker_complete(self, result: WorkerResult) -> None:
894 """Callback when a worker completes.
896 Args:
897 result: Result from the worker
898 """
899 # Unregister from overlap tracking (ENH-143)
900 if self.overlap_detector:
901 self.overlap_detector.unregister_issue(result.issue_id)
902 # Re-queue deferred issues that were waiting on this one
903 self._requeue_deferred_issues()
905 # Handle interrupted workers (not counted as failed) - ENH-036
906 if result.interrupted:
907 self.logger.info(f"{result.issue_id} was interrupted during shutdown (can retry)")
908 self._interrupted_issues.append(result.issue_id)
909 # Don't mark as failed - they can be retried on next run
910 return
912 # Handle issue closure (no merge needed)
913 if result.should_close:
914 # Lazy import to avoid circular dependency
915 from little_loops.issue_lifecycle import close_issue
917 self.logger.info(f"{result.issue_id} should be closed: {result.close_status}")
918 info = self._issue_info_by_id.get(result.issue_id)
919 if info:
920 # TODO(ENH-1686): parallel-path close events not yet live-written
921 if close_issue(
922 info,
923 self.br_config,
924 self.logger,
925 result.close_reason,
926 result.close_status,
927 interceptors=None,
928 ):
929 self.queue.mark_completed(result.issue_id)
930 else:
931 self._worker_errors[result.issue_id] = (
932 f"Close failed: {result.close_reason or 'close error'}"
933 )
934 self.queue.mark_failed(result.issue_id)
935 else:
936 self.logger.warning(f"No issue info found for {result.issue_id}")
937 self._worker_errors[result.issue_id] = "Close failed: no issue info"
938 self.queue.mark_failed(result.issue_id)
939 elif result.success:
940 self.logger.success(
941 f"{result.issue_id} completed in {format_duration(result.duration)}"
942 )
943 if result.was_corrected:
944 self.logger.info(f"{result.issue_id} was auto-corrected during validation")
945 # Log and store corrections for pattern analysis (ENH-010)
946 for correction in result.corrections:
947 self.logger.info(f" Correction: {correction}")
948 if result.corrections:
949 with self._state_lock:
950 self.state.corrections[result.issue_id] = result.corrections
951 if self.parallel_config.use_feature_branches:
952 # Feature branch mode: skip auto-merge, branch stays alive (ENH-665, BUG-2172)
953 self.logger.info(f"{result.issue_id}: feature branch ready — {result.branch_name}")
954 self.queue.mark_completed(result.issue_id)
955 self._complete_issue_lifecycle_if_needed(result.issue_id)
956 branch_state: dict[str, Any] = {
957 "branch_name": result.branch_name,
958 "pushed": False,
959 "pr_url": None,
960 }
961 if self.parallel_config.push_feature_branches:
962 push_result = subprocess.run(
963 [
964 "git",
965 "push",
966 "--force-with-lease",
967 self.parallel_config.remote_name,
968 result.branch_name,
969 ],
970 cwd=self.repo_path,
971 capture_output=True,
972 text=True,
973 timeout=60,
974 )
975 if push_result.returncode == 0:
976 branch_state["pushed"] = True
977 self.logger.info(
978 f"{result.issue_id}: pushed {result.branch_name}"
979 f" to {self.parallel_config.remote_name}"
980 )
981 if self.parallel_config.open_pr_for_feature_branches:
982 self._open_pr_for_branch(result.issue_id, result.branch_name, branch_state)
983 else:
984 self.logger.warning(
985 f"{result.issue_id}: git push failed: {push_result.stderr.strip()}"
986 )
987 with self._state_lock:
988 self._pr_ready_branches[result.issue_id] = branch_state
989 else:
990 self.merge_coordinator.queue_merge(result)
991 # Wait for merge to complete before returning from callback.
992 # This prevents dispatch of next worker while merge is in progress,
993 # avoiding race conditions between worktree creation and merge ops.
994 # (BUG-140: Race condition between worktree creation and merge)
995 self.merge_coordinator.wait_for_completion(timeout=120)
996 if result.issue_id in self.merge_coordinator.merged_ids:
997 self.queue.mark_completed(result.issue_id)
998 self._complete_issue_lifecycle_if_needed(result.issue_id)
999 else:
1000 self._worker_errors[result.issue_id] = (
1001 f"Merge failed: {result.error or 'merge error'}"
1002 )
1003 self.queue.mark_failed(result.issue_id)
1004 else:
1005 self.logger.error(f"{result.issue_id} failed: {result.error}")
1006 self._worker_errors[result.issue_id] = result.error or "Failed"
1007 self.queue.mark_failed(result.issue_id)
1009 # Update timing
1010 with self._state_lock:
1011 self.state.timing[result.issue_id] = {
1012 "total": result.duration,
1013 }
1015 # Clean up stage tracking after callback completes (ENH-262)
1016 # Delay briefly so status reporter can show completion
1017 self.worker_pool.remove_worker_stage(result.issue_id)
1019 # Emit worker completion event for extensions (ENH-921)
1020 if self._event_bus:
1021 self._event_bus.emit(
1022 {
1023 "event": "parallel.worker_completed",
1024 "ts": datetime.now(UTC).isoformat(),
1025 "issue_id": result.issue_id,
1026 "worker_name": result.worktree_path.name,
1027 "status": "success" if result.success else "failure",
1028 "duration_seconds": result.duration,
1029 }
1030 )
1032 def _requeue_deferred_issues(self) -> None:
1033 """Re-queue deferred issues that no longer have overlaps (ENH-143)."""
1034 if not self._deferred_issues:
1035 return
1037 # Check each deferred issue for remaining overlaps
1038 still_deferred = []
1039 for issue in self._deferred_issues:
1040 if self.overlap_detector:
1041 overlap = self.overlap_detector.check_overlap(issue)
1042 if overlap:
1043 # Still has overlaps, keep deferred
1044 still_deferred.append(issue)
1045 else:
1046 # No more overlaps, add back to queue
1047 self.logger.info(f"Re-queuing {issue.issue_id} - no longer overlapping")
1048 self.queue.add(issue)
1050 self._deferred_issues = still_deferred
1052 def _open_pr_for_branch(
1053 self,
1054 issue_id: str,
1055 branch_name: str,
1056 branch_state: dict[str, Any],
1057 ) -> None:
1058 """Open a draft PR for a pushed feature branch using the gh CLI.
1060 Mutates branch_state in place to record pr_url on success.
1061 Degrades gracefully if gh is missing or unauthenticated.
1062 """
1063 try:
1064 auth_result = subprocess.run(
1065 ["gh", "auth", "status"],
1066 capture_output=True,
1067 text=True,
1068 timeout=30,
1069 )
1070 if auth_result.returncode != 0:
1071 self.logger.warning(f"{issue_id}: gh not authenticated, skipping PR creation")
1072 return
1073 issue_info = self._issue_info_by_id.get(issue_id)
1074 pr_title = issue_info.title if issue_info else issue_id
1075 pr_result = subprocess.run(
1076 [
1077 "gh",
1078 "pr",
1079 "create",
1080 "--title",
1081 pr_title,
1082 "--body",
1083 f"Closes {issue_id}",
1084 "--base",
1085 self.parallel_config.base_branch,
1086 "--draft",
1087 "--head",
1088 branch_name,
1089 ],
1090 cwd=self.repo_path,
1091 capture_output=True,
1092 text=True,
1093 timeout=60,
1094 )
1095 if pr_result.returncode == 0:
1096 branch_state["pr_url"] = pr_result.stdout.strip()
1097 self.logger.info(f"{issue_id}: PR opened: {branch_state['pr_url']}")
1098 else:
1099 self.logger.warning(
1100 f"{issue_id}: gh pr create failed: {pr_result.stderr.strip()}"
1101 )
1102 except FileNotFoundError:
1103 self.logger.warning(f"{issue_id}: gh CLI not found, skipping PR creation")
1104 except subprocess.TimeoutExpired:
1105 self.logger.warning(f"{issue_id}: gh pr create timed out")
1107 def _merge_sequential(self, result: WorkerResult) -> None:
1108 """Merge a sequential (P0) result immediately.
1110 Args:
1111 result: Result to merge
1112 """
1113 # Handle closure for sequential issues
1114 if result.should_close:
1115 # Lazy import to avoid circular dependency
1116 from little_loops.issue_lifecycle import close_issue
1118 info = self._issue_info_by_id.get(result.issue_id)
1119 # TODO(ENH-1686): parallel-path close events not yet live-written
1120 if info and close_issue(
1121 info,
1122 self.br_config,
1123 self.logger,
1124 result.close_reason,
1125 result.close_status,
1126 interceptors=None,
1127 ):
1128 self.queue.mark_completed(result.issue_id)
1129 else:
1130 self._worker_errors[result.issue_id] = (
1131 f"Close failed: {result.close_reason or 'close error'}"
1132 )
1133 self.queue.mark_failed(result.issue_id)
1134 return
1136 self.merge_coordinator.queue_merge(result)
1137 # Wait for this specific merge
1138 self.merge_coordinator.wait_for_completion(timeout=60)
1140 if result.issue_id in self.merge_coordinator.merged_ids:
1141 self.queue.mark_completed(result.issue_id)
1142 self._complete_issue_lifecycle_if_needed(result.issue_id)
1143 else:
1144 self._worker_errors[result.issue_id] = f"Merge failed: {result.error or 'merge error'}"
1145 self.queue.mark_failed(result.issue_id)
1147 def _wait_for_completion(self) -> None:
1148 """Wait for all workers and merges to complete."""
1149 self.logger.info("Waiting for workers to complete...")
1151 # Calculate timeout
1152 if self.parallel_config.orchestrator_timeout > 0:
1153 timeout = self.parallel_config.orchestrator_timeout
1154 else:
1155 timeout = self.parallel_config.timeout_per_issue * self.parallel_config.max_workers
1157 start = time.time()
1158 while self.worker_pool.active_count > 0:
1159 if time.time() - start > timeout:
1160 self.logger.warning(f"Timeout waiting for workers after {timeout}s")
1161 self.worker_pool.terminate_all_processes()
1162 break
1163 time.sleep(1.0)
1165 # Wait for merges
1166 self.logger.info("Waiting for pending merges...")
1167 self.merge_coordinator.wait_for_completion(timeout=120)
1169 # Update queue with merge results and complete lifecycle
1170 for issue_id in self.merge_coordinator.merged_ids:
1171 self.queue.mark_completed(issue_id)
1172 self._complete_issue_lifecycle_if_needed(issue_id)
1174 for issue_id, reason in self.merge_coordinator.failed_merges.items():
1175 self._worker_errors[issue_id] = reason or "Merge failed"
1176 self.queue.mark_failed(issue_id)
1178 def _report_results(self, start_time: float) -> None:
1179 """Report processing results.
1181 Args:
1182 start_time: When processing started
1183 """
1184 total_time = time.time() - start_time
1185 self._execution_duration = total_time
1187 self.logger.info("")
1188 self.logger.info("=" * 60)
1189 if self.wave_label:
1190 self.logger.info(f"{self.wave_label.upper()} PROCESSING COMPLETE")
1191 else:
1192 self.logger.info("PARALLEL ISSUE PROCESSING COMPLETE")
1193 self.logger.info("=" * 60)
1194 self.logger.info("")
1195 self.logger.timing(f"Total time: {format_duration(total_time)}")
1196 self.logger.info(f"Completed: {self.queue.completed_count}")
1197 self.logger.info(f"Failed: {self.queue.failed_count}")
1198 if self._interrupted_issues:
1199 self.logger.info(f"Interrupted: {len(self._interrupted_issues)}")
1201 with self._state_lock:
1202 timing_snapshot = dict(self.state.timing)
1203 corrections_snapshot = dict(self.state.corrections)
1205 if self.queue.completed_count > 0:
1206 total_issue_time = sum(t.get("total", 0) for t in timing_snapshot.values())
1207 if total_issue_time > 0:
1208 speedup = total_issue_time / total_time
1209 self.logger.info(f"Estimated speedup: {speedup:.2f}x")
1211 if self.queue.failed_ids:
1212 self.logger.info("")
1213 self.logger.warning("Failed issues:")
1214 for issue_id in self.queue.failed_ids:
1215 self.logger.warning(f" - {issue_id}")
1217 # Report interrupted issues separately (ENH-036)
1218 if self._interrupted_issues:
1219 self.logger.info("")
1220 self.logger.info(f"Interrupted: {len(self._interrupted_issues)} (can retry)")
1221 for issue_id in self._interrupted_issues:
1222 self.logger.info(f" - {issue_id}")
1224 # Report feature branches with actual per-branch state (ENH-665, BUG-2172)
1225 if self._pr_ready_branches:
1226 self.logger.info("")
1227 self.logger.info(f"Feature branches: {len(self._pr_ready_branches)} branch(es)")
1228 for issue_id, state in self._pr_ready_branches.items():
1229 branch = state["branch_name"]
1230 if state.get("pr_url"):
1231 self.logger.info(
1232 f" - {issue_id}: {branch} — pushed + PR opened: {state['pr_url']}"
1233 )
1234 elif state.get("pushed"):
1235 self.logger.info(f" - {issue_id}: {branch} — pushed (PR skipped)")
1236 else:
1237 self.logger.info(f" - {issue_id}: {branch} — local-only branch retained")
1239 # Report correction statistics for quality tracking (ENH-010)
1240 if corrections_snapshot:
1241 total_corrected = len(corrections_snapshot)
1242 total_issues = self.queue.completed_count + self.queue.failed_count
1243 correction_rate = (total_corrected / total_issues * 100) if total_issues > 0 else 0
1244 self.logger.info("")
1245 self.logger.info(
1246 f"Auto-corrections: {total_corrected}/{total_issues} ({correction_rate:.1f}%)"
1247 )
1249 # Group corrections by category (ENH-010 fourth fix)
1250 from collections import Counter, defaultdict
1252 all_corrections: list[str] = []
1253 by_category: dict[str, int] = defaultdict(int)
1254 for corrections in corrections_snapshot.values():
1255 all_corrections.extend(corrections)
1256 for correction in corrections:
1257 # Extract category from [category] prefix if present
1258 if correction.startswith("[") and "]" in correction:
1259 category = correction[1 : correction.index("]")]
1260 by_category[category] += 1
1261 else:
1262 by_category["uncategorized"] += 1
1264 # Log corrections by type/category
1265 if by_category:
1266 self.logger.info("Corrections by type:")
1267 for category, count in sorted(by_category.items(), key=lambda x: -x[1]):
1268 self.logger.info(f" - {category}: {count}")
1270 # Log most common individual corrections
1271 if all_corrections:
1272 common = Counter(all_corrections).most_common(3)
1273 self.logger.info("Most common corrections:")
1274 for correction, count in common:
1275 # Truncate long correction descriptions
1276 display = correction[:60] + "..." if len(correction) > 60 else correction
1277 self.logger.info(f" - {display}: {count}")
1279 # Report stash pop warnings (local changes need manual recovery)
1280 stash_warnings = self.merge_coordinator.stash_pop_failures
1281 if stash_warnings:
1282 self.logger.info("")
1283 self.logger.warning("Stash recovery warnings (local changes need manual restoration):")
1284 for issue_id, message in stash_warnings.items():
1285 self.logger.warning(f" - {issue_id}: {message}")
1286 self.logger.warning("")
1287 self.logger.warning(
1288 "To recover: Run 'git stash list' to find your changes, "
1289 "then 'git stash pop' or 'git stash apply stash@{N}'"
1290 )
1292 def _complete_issue_lifecycle_if_needed(self, issue_id: str) -> bool:
1293 """Complete issue lifecycle by writing ``status: done`` to frontmatter.
1295 Args:
1296 issue_id: ID of the issue to complete
1298 Returns:
1299 True if lifecycle was completed (or already complete), False on error
1300 """
1301 # TODO(ENH-1686): parallel-path close events not yet live-written
1302 info = self._issue_info_by_id.get(issue_id)
1303 if not info:
1304 self.logger.warning(f"No issue info found for {issue_id}")
1305 return False
1307 original_path = info.path
1309 if not original_path.exists():
1310 return True
1312 self.logger.info(f"Completing lifecycle for {issue_id} (frontmatter status update)")
1314 try:
1315 content = original_path.read_text()
1317 # Add resolution section if not already present
1318 if "## Resolution" not in content:
1319 action = self.br_config.get_category_action(info.issue_type)
1320 resolution = f"""
1322---
1324## Resolution
1326- **Action**: {action}
1327- **Completed**: {datetime.now(UTC).strftime("%Y-%m-%d")}
1328- **Status**: Completed (parallel merge fallback)
1329- **Implementation**: Merged from parallel worker branch
1331### Changes Made
1332- See git history for implementation details
1334### Verification Results
1335- Work verification passed before merge
1337### Commits
1338- See `git log --oneline` for merge commit details
1339"""
1340 content += resolution
1342 content = update_frontmatter(
1343 content,
1344 {
1345 "status": "done",
1346 "completed_at": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
1347 },
1348 )
1349 original_path.write_text(content)
1350 append_session_log_entry(original_path, "ll-parallel")
1352 # Stage and commit
1353 self._git_lock.run(
1354 ["add", "-A"],
1355 cwd=self.repo_path,
1356 )
1358 action = self.br_config.get_category_action(info.issue_type)
1359 commit_msg = f"""{action}({info.issue_type}): complete {issue_id} lifecycle
1361Parallel merge fallback - status: done written to frontmatter.
1363Issue: {issue_id}
1364Type: {info.issue_type}
1365Title: {info.title}
1366"""
1367 commit_result = self._git_lock.run(
1368 ["commit", "-m", commit_msg],
1369 cwd=self.repo_path,
1370 )
1372 if commit_result.returncode != 0:
1373 if "nothing to commit" not in commit_result.stdout.lower():
1374 self.logger.warning(f"git commit failed: {commit_result.stderr}")
1375 else:
1376 commit_hash_match = re.search(r"\[[\w-]+\s+([a-f0-9]+)\]", commit_result.stdout)
1377 if commit_hash_match:
1378 self.logger.success(
1379 f"Completed lifecycle for {issue_id}: {commit_hash_match.group(1)}"
1380 )
1381 else:
1382 self.logger.success(f"Completed lifecycle for {issue_id}")
1384 return True
1386 except Exception as e:
1387 self.logger.error(f"Failed to complete lifecycle for {issue_id}: {e}")
1388 return False
1390 def _cleanup(self) -> None:
1391 """Clean up resources."""
1392 self.logger.info("Cleaning up...")
1394 # Save final state (force=True bypasses throttle to ensure shutdown state is persisted)
1395 self._save_state(force=True)
1397 # Shutdown components
1398 self.worker_pool.shutdown(wait=True)
1399 self.merge_coordinator.shutdown(wait=True, timeout=30)
1401 # Flush transports regardless of interrupt state so events are not lost.
1402 if self._event_bus is not None:
1403 self._event_bus.close_transports()
1405 # Clean up worktrees if not interrupted
1406 if not self._shutdown_requested:
1407 self.worker_pool.cleanup_all_worktrees()