Coverage for little_loops / parallel / orchestrator.py: 9%
610 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -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.issue_parser import IssueInfo
24from little_loops.logger import Logger, format_duration
25from little_loops.parallel.git_lock import GitLock
26from little_loops.parallel.merge_coordinator import MergeCoordinator
27from little_loops.parallel.overlap_detector import OverlapDetector
28from little_loops.parallel.priority_queue import IssuePriorityQueue
29from little_loops.parallel.types import (
30 OrchestratorState,
31 ParallelConfig,
32 PendingWorktreeInfo,
33 WorkerResult,
34)
35from little_loops.parallel.worker_pool import WorkerPool
36from little_loops.session_log import append_session_log_entry
38if TYPE_CHECKING:
39 from little_loops.config import BRConfig
42class ParallelOrchestrator:
43 """Main controller for parallel issue processing.
45 Coordinates:
46 - Issue scanning and prioritization
47 - Worker dispatch (P0 sequential, P1-P5 parallel)
48 - Merge coordination
49 - State persistence for resume capability
50 - Graceful shutdown on signals
52 Example:
53 >>> from little_loops.config import BRConfig
54 >>> from little_loops.parallel import ParallelConfig, ParallelOrchestrator
55 >>> br_config = BRConfig(Path.cwd())
56 >>> parallel_config = ParallelConfig(max_workers=2)
57 >>> orchestrator = ParallelOrchestrator(parallel_config, br_config)
58 >>> exit_code = orchestrator.run()
59 """
61 def __init__(
62 self,
63 parallel_config: ParallelConfig,
64 br_config: BRConfig,
65 repo_path: Path | None = None,
66 verbose: bool = True,
67 wave_label: str | None = None,
68 event_bus: EventBus | None = None,
69 ) -> None:
70 """Initialize the orchestrator.
72 Args:
73 parallel_config: Parallel processing configuration
74 br_config: Project configuration
75 repo_path: Path to the git repository (default: current directory)
76 verbose: Whether to output progress messages
77 wave_label: Optional label for wave-based execution (e.g., "Wave 1")
78 event_bus: Optional EventBus for emitting worker completion events
79 """
80 self.parallel_config = parallel_config
81 self.br_config = br_config
82 self.repo_path = repo_path or Path.cwd()
83 self.logger = Logger(verbose=verbose)
84 self.wave_label = wave_label
85 self._event_bus = event_bus
86 self._execution_duration: float = 0.0
88 # Create shared git lock for serializing main repo operations
89 # This prevents index.lock race conditions between workers and merge coordinator
90 self._git_lock = GitLock(self.logger)
92 # Initialize components with shared git lock
93 self.queue = IssuePriorityQueue()
94 self.worker_pool = WorkerPool(
95 parallel_config, br_config, self.logger, self.repo_path, self._git_lock
96 )
97 self.merge_coordinator = MergeCoordinator(
98 parallel_config, self.logger, self.repo_path, self._git_lock
99 )
101 # State management
102 self.state = OrchestratorState()
103 self._state_lock = threading.Lock()
104 self._shutdown_requested = False
105 self._original_sigint: Any = None
106 self._original_sigterm: Any = None
108 # Track issue info for lifecycle completion after merge
109 self._issue_info_by_id: dict[str, IssueInfo] = {}
110 # Track interrupted issues separately from failures (ENH-036)
111 self._interrupted_issues: list[str] = []
112 # Track PR-ready branches when use_feature_branches=True (ENH-665)
113 self._pr_ready_branches: dict[str, str] = {} # issue_id -> branch_name
115 # Overlap detection (ENH-143)
116 self.overlap_detector: OverlapDetector | None = (
117 OverlapDetector(config=br_config.dependency_mapping)
118 if parallel_config.overlap_detection
119 else None
120 )
121 # Track deferred issues for re-check after active issues complete
122 self._deferred_issues: list[IssueInfo] = []
123 # Track last status report time for progress visibility (ENH-262)
124 self._last_status_time: float = 0.0
125 self._last_status_line: str = ""
126 # Track last state save time to throttle disk writes (ENH-485)
127 self._last_save_time: float = 0.0
129 @property
130 def execution_duration(self) -> float:
131 """Return the total execution duration in seconds."""
132 return self._execution_duration
134 def run(self) -> int:
135 """Run the parallel issue processor.
137 Returns:
138 Exit code (0 = success, 1 = failure)
139 """
140 try:
141 self._setup_signal_handlers()
142 self._ensure_gitignore_entries()
144 # Check for pending work from previous runs (unless clean start)
145 if not self.parallel_config.clean_start:
146 pending_worktrees = self._check_pending_worktrees()
148 # Handle pending worktrees based on flags
149 pending_with_work = [p for p in pending_worktrees if p.has_pending_work]
150 if pending_with_work:
151 if self.parallel_config.merge_pending:
152 self._merge_pending_worktrees(pending_worktrees)
153 elif not self.parallel_config.ignore_pending:
154 # Default behavior: just report (cleanup happens below)
155 self.logger.info(
156 "Continuing with cleanup (use --merge-pending to merge)..."
157 )
159 self._cleanup_orphaned_worktrees()
160 self._load_state()
162 if self.parallel_config.dry_run:
163 return self._dry_run()
165 return self._execute()
167 except KeyboardInterrupt:
168 self.logger.warning("Interrupted by user")
169 return 1
170 except Exception as e:
171 self.logger.error(f"Fatal error: {e}")
172 return 1
173 finally:
174 self._cleanup()
175 self._restore_signal_handlers()
177 def _setup_signal_handlers(self) -> None:
178 """Setup signal handlers for graceful shutdown."""
179 self._original_sigint = signal.signal(signal.SIGINT, self._signal_handler)
180 self._original_sigterm = signal.signal(signal.SIGTERM, self._signal_handler)
182 def _restore_signal_handlers(self) -> None:
183 """Restore original signal handlers."""
184 if self._original_sigint is not None:
185 signal.signal(signal.SIGINT, self._original_sigint)
186 if self._original_sigterm is not None:
187 signal.signal(signal.SIGTERM, self._original_sigterm)
189 def _ensure_gitignore_entries(self) -> None:
190 """Ensure .gitignore has entries for parallel processing artifacts.
192 Adds entries for:
193 - .parallel-manage-state.json (state file)
194 - .worktrees/ (git worktree directory)
196 This prevents these files from being tracked by git, which would cause
197 conflicts during merge operations (state file is continuously updated).
198 """
199 gitignore_path = self.repo_path / ".gitignore"
200 required_entries = [
201 ".parallel-manage-state.json",
202 ".worktrees/",
203 ]
205 existing_content = ""
206 if gitignore_path.exists():
207 existing_content = gitignore_path.read_text()
209 # Check which entries are missing
210 missing_entries = []
211 for entry in required_entries:
212 # Check for exact match or pattern that would cover it
213 if entry not in existing_content:
214 missing_entries.append(entry)
216 if not missing_entries:
217 return
219 # Append missing entries
220 addition = "\n# ll-parallel artifacts\n"
221 for entry in missing_entries:
222 addition += f"{entry}\n"
224 # Ensure file ends with newline before adding
225 if existing_content and not existing_content.endswith("\n"):
226 addition = "\n" + addition
228 gitignore_path.write_text(existing_content + addition)
229 self.logger.info(f"Added {len(missing_entries)} entries to .gitignore")
231 def _cleanup_orphaned_worktrees(self) -> None:
232 """Clean up worktrees from previous interrupted runs.
234 Scans the worktree base directory and removes any worktrees that are
235 not from the current session. This handles cases where a previous run
236 was interrupted (Ctrl+C) and worktrees were not cleaned up.
237 """
238 worktree_base = self.repo_path / self.parallel_config.worktree_base
239 if not worktree_base.exists():
240 return
242 # Get list of worktree directories, skipping those owned by live processes (BUG-579)
243 orphaned = []
244 for item in worktree_base.iterdir():
245 if item.is_dir() and item.name.startswith("worker-"):
246 # Check for a .ll-session-<pid> marker left by an active orchestrator
247 owned_by_live = False
248 for marker in item.glob(".ll-session-*"):
249 try:
250 pid = int(marker.name.split("-")[-1])
251 os.kill(pid, 0) # Signal 0: check if process exists
252 owned_by_live = True
253 break
254 except (ProcessLookupError, ValueError):
255 pass
256 except PermissionError:
257 owned_by_live = True # Process exists, no permission to signal
258 break
259 if owned_by_live:
260 self.logger.info(f"Skipping {item.name}: owned by running process")
261 continue
262 orphaned.append(item)
264 if not orphaned:
265 return
267 self.logger.info(f"Cleaning up {len(orphaned)} orphaned worktree(s) from previous run")
269 for worktree_path in orphaned:
270 try:
271 # Try git worktree remove first
272 self._git_lock.run(
273 ["worktree", "remove", "--force", str(worktree_path)],
274 cwd=self.repo_path,
275 timeout=30,
276 )
278 # If git worktree remove failed, force delete the directory
279 if worktree_path.exists():
280 shutil.rmtree(worktree_path, ignore_errors=True)
282 # Try to delete the associated branch using the actual branch name
283 branch_result = subprocess.run(
284 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
285 cwd=worktree_path,
286 capture_output=True,
287 text=True,
288 )
289 branch_name = (
290 branch_result.stdout.strip() if branch_result.returncode == 0 else None
291 )
292 if branch_name and branch_name.startswith("parallel/"):
293 self._git_lock.run(
294 ["branch", "-D", branch_name],
295 cwd=self.repo_path,
296 timeout=10,
297 )
298 except Exception as e:
299 self.logger.warning(f"Failed to clean up {worktree_path.name}: {e}")
301 # Also prune git worktree references
302 self._git_lock.run(
303 ["worktree", "prune"],
304 cwd=self.repo_path,
305 timeout=30,
306 )
308 def _inspect_worktree(self, worktree_path: Path) -> PendingWorktreeInfo | None:
309 """Inspect a worktree to determine its status.
311 Args:
312 worktree_path: Path to the worktree directory
314 Returns:
315 PendingWorktreeInfo if inspection succeeded, None if failed
316 """
317 try:
318 # Read actual branch name from worktree via rev-parse
319 branch_result = subprocess.run(
320 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
321 cwd=worktree_path,
322 capture_output=True,
323 text=True,
324 )
325 if branch_result.returncode == 0:
326 branch_name = branch_result.stdout.strip()
327 else:
328 # Fall back to string derivation if rev-parse fails
329 branch_name = worktree_path.name.replace("worker-", "parallel/")
331 # Extract issue ID (e.g., bug-045 -> BUG-045)
332 # Pattern: worker-<issue-id>-<timestamp>
333 match = re.match(r"worker-([a-z]+-\d+)-\d{8}-\d{6}", worktree_path.name)
334 issue_id = match.group(1).upper() if match else worktree_path.name
336 # Check commits ahead of main
337 result = self._git_lock.run(
338 ["rev-list", "--count", f"{self.parallel_config.base_branch}..{branch_name}"],
339 cwd=self.repo_path,
340 timeout=10,
341 )
342 commits_ahead = int(result.stdout.strip()) if result.returncode == 0 else 0
344 # Check for uncommitted changes in worktree
345 result = self._git_lock.run(
346 ["status", "--porcelain"],
347 cwd=worktree_path,
348 timeout=10,
349 )
350 changed_files = []
351 has_uncommitted = False
352 if result.returncode == 0 and result.stdout.strip():
353 has_uncommitted = True
354 changed_files = [line[3:] for line in result.stdout.strip().split("\n") if line]
356 return PendingWorktreeInfo(
357 worktree_path=worktree_path,
358 branch_name=branch_name,
359 issue_id=issue_id,
360 commits_ahead=commits_ahead,
361 has_uncommitted_changes=has_uncommitted,
362 changed_files=changed_files,
363 )
364 except Exception as e:
365 self.logger.warning(f"Failed to inspect worktree {worktree_path.name}: {e}")
366 return None
368 def _check_pending_worktrees(self) -> list[PendingWorktreeInfo]:
369 """Check for pending worktrees from previous runs and report status.
371 Returns:
372 List of pending worktree information
373 """
374 worktree_base = self.repo_path / self.parallel_config.worktree_base
375 if not worktree_base.exists():
376 return []
378 # Find all worker directories
379 worktrees = [
380 item
381 for item in worktree_base.iterdir()
382 if item.is_dir() and item.name.startswith("worker-")
383 ]
385 if not worktrees:
386 return []
388 self.logger.info("Checking for pending work from previous runs...")
390 # Inspect each worktree
391 pending_info: list[PendingWorktreeInfo] = []
392 for worktree_path in worktrees:
393 info = self._inspect_worktree(worktree_path)
394 if info:
395 pending_info.append(info)
397 # Report findings
398 with_work = [p for p in pending_info if p.has_pending_work]
399 if with_work:
400 self.logger.warning(f"Found {len(with_work)} worktree(s) with pending work:")
401 for info in with_work:
402 status_parts = []
403 if info.commits_ahead > 0:
404 status_parts.append(f"{info.commits_ahead} commit(s) ahead of main")
405 if info.has_uncommitted_changes:
406 status_parts.append(f"{len(info.changed_files)} uncommitted file(s)")
407 status = ", ".join(status_parts)
408 self.logger.warning(f" - {info.worktree_path.name}: {info.issue_id} ({status})")
410 self.logger.info("")
411 self.logger.info("Options:")
412 self.logger.info(" --merge-pending Attempt to merge pending work before continuing")
413 self.logger.info(" --clean-start Remove all worktrees and start fresh")
414 self.logger.info(
415 " --ignore-pending Continue without action (worktrees will be cleaned up)"
416 )
417 elif pending_info:
418 self.logger.info(f"Found {len(pending_info)} orphaned worktree(s) with no pending work")
420 return pending_info
422 def _merge_pending_worktrees(self, pending: list[PendingWorktreeInfo]) -> None:
423 """Attempt to merge pending worktrees from previous runs.
425 Args:
426 pending: List of pending worktree information
427 """
428 with_work = [p for p in pending if p.has_pending_work]
429 if not with_work:
430 return
432 self.logger.info(f"Attempting to merge {len(with_work)} pending worktree(s)...")
434 for info in with_work:
435 try:
436 # If there are uncommitted changes, commit them first
437 if info.has_uncommitted_changes:
438 self.logger.info(f" Committing uncommitted changes in {info.issue_id}...")
439 self._git_lock.run(
440 ["add", "-A"],
441 cwd=info.worktree_path,
442 timeout=30,
443 )
444 self._git_lock.run(
445 [
446 "commit",
447 "-m",
448 f"WIP: Auto-commit from interrupted session for {info.issue_id}",
449 ],
450 cwd=info.worktree_path,
451 timeout=30,
452 )
454 # Attempt merge
455 self.logger.info(f" Merging {info.issue_id} ({info.branch_name})...")
456 result = self._git_lock.run(
457 [
458 "merge",
459 "--no-ff",
460 info.branch_name,
461 "-m",
462 f"Merge pending work for {info.issue_id}",
463 ],
464 cwd=self.repo_path,
465 timeout=60,
466 )
468 if result.returncode == 0:
469 self.logger.success(f" Successfully merged {info.issue_id}")
470 # Clean up the worktree after successful merge
471 self._git_lock.run(
472 ["worktree", "remove", "--force", str(info.worktree_path)],
473 cwd=self.repo_path,
474 timeout=30,
475 )
476 self._git_lock.run(
477 ["branch", "-D", info.branch_name],
478 cwd=self.repo_path,
479 timeout=10,
480 )
481 else:
482 self.logger.warning(f" Failed to merge {info.issue_id}: {result.stderr}")
483 # Abort the merge if it failed
484 self._git_lock.run(
485 ["merge", "--abort"],
486 cwd=self.repo_path,
487 timeout=10,
488 )
490 except Exception as e:
491 self.logger.warning(f" Error merging {info.issue_id}: {e}")
493 def _signal_handler(self, signum: int, frame: object) -> None:
494 """Handle shutdown signals gracefully."""
495 self._shutdown_requested = True
496 # Propagate to worker pool for interrupted worker detection (ENH-036)
497 self.worker_pool.set_shutdown_requested(True)
498 self.logger.warning(f"Received signal {signum}, shutting down gracefully...")
500 def _load_state(self) -> None:
501 """Load state from file for resume capability."""
502 if self.parallel_config.clean_start:
503 self.state.started_at = datetime.now().isoformat()
504 return
505 state_file = self.repo_path / self.parallel_config.state_file
506 if not state_file.exists():
507 self.state.started_at = datetime.now().isoformat()
508 return
510 try:
511 data = json.loads(state_file.read_text())
512 self.state = OrchestratorState.from_dict(data)
514 # Restore queue state
515 self.queue.load_completed(self.state.completed_issues)
516 self.queue.load_failed(self.state.failed_issues.keys())
518 self.logger.info(
519 f"Resumed from previous state: "
520 f"{len(self.state.completed_issues)} completed, "
521 f"{len(self.state.failed_issues)} failed"
522 )
523 except Exception as e:
524 self.logger.warning(f"Could not load state: {e}")
525 self.state.started_at = datetime.now().isoformat()
527 def _save_state(self, force: bool = False) -> None:
528 """Save current state to file using an atomic write.
530 Writes are throttled to at most once every 5 seconds to reduce filesystem I/O
531 during high-frequency loop ticks (e.g., merge-waiting phase). Pass force=True
532 to bypass the throttle, e.g., on shutdown.
533 """
534 now = time.time()
535 if not force and now - self._last_save_time < 5.0:
536 return
537 self._last_save_time = now
538 with self._state_lock:
539 self.state.last_checkpoint = datetime.now().isoformat()
540 self.state.completed_issues = self.queue.completed_ids
541 self.state.failed_issues = dict.fromkeys(self.queue.failed_ids, "Failed")
542 self.state.in_progress_issues = self.queue.in_progress_ids
544 state_file = self.repo_path / self.parallel_config.state_file
545 data = json.dumps(self.state.to_dict(), indent=2)
546 tmp_fd, tmp_path = tempfile.mkstemp(dir=state_file.parent, suffix=".tmp")
547 try:
548 with os.fdopen(tmp_fd, "w") as f:
549 f.write(data)
550 os.replace(tmp_path, state_file)
551 except Exception:
552 os.unlink(tmp_path)
553 raise
555 def _cleanup_state(self) -> None:
556 """Remove state file on successful completion."""
557 state_file = self.repo_path / self.parallel_config.state_file
558 if state_file.exists():
559 state_file.unlink()
561 def _dry_run(self) -> int:
562 """Preview what would be processed without executing.
564 Returns:
565 Exit code (always 0 for dry run)
566 """
567 issues = self._scan_issues()
569 self.logger.info("=" * 60)
570 self.logger.info("DRY RUN - No changes will be made")
571 self.logger.info("=" * 60)
572 self.logger.info("")
574 if not issues:
575 self.logger.info("No issues found matching criteria")
576 return 0
578 self.logger.info(f"Found {len(issues)} issues to process:")
579 self.logger.info("")
581 # Group by priority
582 by_priority: dict[str, list[IssueInfo]] = {}
583 for issue in issues:
584 by_priority.setdefault(issue.priority, []).append(issue)
586 for priority in IssuePriorityQueue.DEFAULT_PRIORITIES:
587 if priority not in by_priority:
588 continue
590 priority_issues = by_priority[priority]
591 self.logger.info(f" {priority} ({len(priority_issues)} issues):")
592 for issue in priority_issues:
593 mode = (
594 "sequential"
595 if priority == "P0" and self.parallel_config.p0_sequential
596 else "parallel"
597 )
598 self.logger.info(f" - {issue.issue_id}: {issue.title} [{mode}]")
600 self.logger.info("")
601 self.logger.info("Configuration:")
602 self.logger.info(f" Workers: {self.parallel_config.max_workers}")
603 self.logger.info(f" P0 Sequential: {self.parallel_config.p0_sequential}")
604 self.logger.info(f" Max Issues: {self.parallel_config.max_issues or 'unlimited'}")
605 self.logger.info(f" Command Prefix: {self.parallel_config.command_prefix}")
607 return 0
609 def _maybe_report_status(self) -> None:
610 """Report status if enough time has elapsed since last report.
612 Reports every 5 seconds during active processing for progress visibility (ENH-262).
613 Suppresses duplicate lines when nothing has changed.
614 """
615 now = time.time()
616 # Report every 5 seconds
617 if now - self._last_status_time < 5.0:
618 return
620 self._last_status_time = now
622 # Build status line
623 parts = []
625 # Add wave label if present
626 if self.wave_label:
627 parts.append(f"{self.wave_label}")
629 # Get queue counts
630 in_progress = len(self.queue.in_progress_ids)
631 completed = self.queue.completed_count
632 failed = self.queue.failed_count
633 pending_merge = self.merge_coordinator.pending_count
635 parts.append(f"Active: {in_progress}")
636 parts.append(f"Done: {completed}")
637 if failed > 0:
638 parts.append(f"Failed: {failed}")
639 if pending_merge > 0:
640 parts.append(f"Merging: {pending_merge}")
642 # Build status line
643 status = " | ".join(parts)
645 # Get active worker stages
646 active_stages = self.worker_pool.get_active_stages()
648 # Add worker details if any are active
649 if active_stages:
650 # Group by stage
651 by_stage: dict[str, list[str]] = {}
652 for issue_id, worker_stage in active_stages.items():
653 stage_name = worker_stage.value.title()
654 by_stage.setdefault(stage_name, []).append(issue_id)
656 stage_parts = []
657 for stage_name in ["Validating", "Implementing", "Verifying", "Merging"]:
658 if stage_name in by_stage:
659 issue_ids = ", ".join(by_stage[stage_name])
660 stage_parts.append(f"{stage_name}: [{issue_ids}]")
662 if stage_parts:
663 status += " | " + " | ".join(stage_parts)
665 # Skip if nothing changed since last report
666 if status == self._last_status_line:
667 return
668 self._last_status_line = status
670 # Log with gray color to distinguish from normal logs
671 if self.logger.use_color:
672 color = self.logger.GRAY
673 ts = self.logger._timestamp()
674 print(f"{color}[{ts}]{self.logger.RESET} {status}", flush=True)
675 else:
676 self.logger.info(status)
678 def _execute(self) -> int:
679 """Execute parallel issue processing.
681 Returns:
682 Exit code (0 = success, 1 = failure)
683 """
684 start_time = time.time()
686 # Scan and queue issues
687 issues = self._scan_issues()
688 if not issues:
689 self.logger.info("No issues to process")
690 return 0
692 # Store issue info for lifecycle completion after merge
693 for issue in issues:
694 self._issue_info_by_id[issue.issue_id] = issue
696 added = self.queue.add_many(issues)
697 self.logger.info(f"Queued {added} issues for processing")
699 # Start components
700 self.worker_pool.start()
701 self.merge_coordinator.start()
703 # Process issues
704 issues_processed = 0
705 max_issues = self.parallel_config.max_issues or float("inf")
707 while not self._shutdown_requested:
708 # Check if done
709 if self.queue.empty() and self.worker_pool.active_count == 0:
710 # Wait for pending merges
711 if self.merge_coordinator.pending_count == 0:
712 break
714 # Check max issues limit
715 if issues_processed >= max_issues:
716 self.logger.info(f"Reached max issues limit ({max_issues})")
717 break
719 # Get next issue if workers available
720 if self.worker_pool.active_count < self.parallel_config.max_workers:
721 queued = self.queue.get(block=False)
722 if queued:
723 issue = queued.issue_info
725 # P0 sequential processing
726 if issue.priority == "P0" and self.parallel_config.p0_sequential:
727 self._process_sequential(issue)
728 else:
729 self._process_parallel(issue)
731 issues_processed += 1
733 # Save state periodically
734 self._save_state()
736 # Report status periodically for progress visibility (ENH-262)
737 self._maybe_report_status()
739 # Small sleep to prevent busy loop
740 time.sleep(0.1)
742 # Wait for completion
743 self._wait_for_completion()
745 # Report results
746 self._report_results(start_time)
748 # Cleanup state on success
749 if not self._shutdown_requested and self.queue.failed_count == 0:
750 self._cleanup_state()
752 return 0 if self.queue.failed_count == 0 else 1
754 def _scan_issues(self) -> list[IssueInfo]:
755 """Scan for issues matching criteria.
757 Returns:
758 List of issues sorted by priority
759 """
760 # Combine skip_ids from state and config
761 skip_ids = set(self.state.completed_issues) | set(self.state.failed_issues.keys())
762 if self.parallel_config.skip_ids:
763 skip_ids |= self.parallel_config.skip_ids
765 issues = IssuePriorityQueue.scan_issues(
766 self.br_config,
767 priority_filter=list(self.parallel_config.priority_filter),
768 skip_ids=skip_ids,
769 only_ids=self.parallel_config.only_ids,
770 type_prefixes=self.parallel_config.type_prefixes,
771 )
773 # Apply max issues limit
774 if self.parallel_config.max_issues > 0:
775 issues = issues[: self.parallel_config.max_issues]
777 return issues
779 def _process_sequential(self, issue: IssueInfo) -> None:
780 """Process an issue sequentially (blocking).
782 Args:
783 issue: Issue to process
784 """
785 self.logger.info(f"Processing {issue.issue_id} sequentially (P0)")
787 # Wait for any parallel work to finish
788 while self.worker_pool.active_count > 0:
789 time.sleep(0.5)
791 # Process in main repo (no worktree isolation needed)
792 # Note: No callback here - _merge_sequential handles the result explicitly
793 # to avoid double-processing (callback would also queue merge/close)
794 future = self.worker_pool.submit(issue)
796 # Wait for completion
797 try:
798 result = future.result(timeout=self.parallel_config.timeout_per_issue)
799 if result.success:
800 # Merge immediately for P0
801 self._merge_sequential(result)
802 except Exception as e:
803 self.logger.error(f"Sequential processing failed: {e}")
804 self.queue.mark_failed(issue.issue_id)
806 def _process_parallel(self, issue: IssueInfo) -> None:
807 """Process an issue in parallel (non-blocking).
809 Args:
810 issue: Issue to process
811 """
812 # Check for overlaps if enabled (ENH-143)
813 if self.overlap_detector:
814 overlap = self.overlap_detector.check_overlap(issue)
815 if overlap:
816 if self.parallel_config.serialize_overlapping:
817 self.logger.warning(
818 f"Deferring {issue.issue_id} - overlaps with {overlap.overlapping_issues}"
819 )
820 # Track for re-check when active issues complete
821 self._deferred_issues.append(issue)
822 return
823 else:
824 self.logger.warning(
825 f"Warning: {issue.issue_id} may conflict with {overlap.overlapping_issues}"
826 )
828 # Register as active before dispatch
829 self.overlap_detector.register_issue(issue)
831 self.logger.info(f"Dispatching {issue.issue_id} to worker pool")
832 self.worker_pool.submit(issue, self._on_worker_complete)
834 def _on_worker_complete(self, result: WorkerResult) -> None:
835 """Callback when a worker completes.
837 Args:
838 result: Result from the worker
839 """
840 # Unregister from overlap tracking (ENH-143)
841 if self.overlap_detector:
842 self.overlap_detector.unregister_issue(result.issue_id)
843 # Re-queue deferred issues that were waiting on this one
844 self._requeue_deferred_issues()
846 # Handle interrupted workers (not counted as failed) - ENH-036
847 if result.interrupted:
848 self.logger.info(f"{result.issue_id} was interrupted during shutdown (can retry)")
849 self._interrupted_issues.append(result.issue_id)
850 # Don't mark as failed - they can be retried on next run
851 return
853 # Handle issue closure (no merge needed)
854 if result.should_close:
855 # Lazy import to avoid circular dependency
856 from little_loops.issue_lifecycle import close_issue
858 self.logger.info(f"{result.issue_id} should be closed: {result.close_status}")
859 info = self._issue_info_by_id.get(result.issue_id)
860 if info:
861 if close_issue(
862 info,
863 self.br_config,
864 self.logger,
865 result.close_reason,
866 result.close_status,
867 interceptors=None,
868 ):
869 self.queue.mark_completed(result.issue_id)
870 else:
871 self.queue.mark_failed(result.issue_id)
872 else:
873 self.logger.warning(f"No issue info found for {result.issue_id}")
874 self.queue.mark_failed(result.issue_id)
875 elif result.success:
876 self.logger.success(
877 f"{result.issue_id} completed in {format_duration(result.duration)}"
878 )
879 if result.was_corrected:
880 self.logger.info(f"{result.issue_id} was auto-corrected during validation")
881 # Log and store corrections for pattern analysis (ENH-010)
882 for correction in result.corrections:
883 self.logger.info(f" Correction: {correction}")
884 if result.corrections:
885 with self._state_lock:
886 self.state.corrections[result.issue_id] = result.corrections
887 if self.parallel_config.use_feature_branches:
888 # Feature branch mode: skip auto-merge, branch stays alive for a PR (ENH-665)
889 self.logger.info(f"{result.issue_id}: feature branch ready — {result.branch_name}")
890 self.queue.mark_completed(result.issue_id)
891 self._complete_issue_lifecycle_if_needed(result.issue_id)
892 with self._state_lock:
893 self._pr_ready_branches[result.issue_id] = result.branch_name
894 else:
895 self.merge_coordinator.queue_merge(result)
896 # Wait for merge to complete before returning from callback.
897 # This prevents dispatch of next worker while merge is in progress,
898 # avoiding race conditions between worktree creation and merge ops.
899 # (BUG-140: Race condition between worktree creation and merge)
900 self.merge_coordinator.wait_for_completion(timeout=120)
901 if result.issue_id in self.merge_coordinator.merged_ids:
902 self.queue.mark_completed(result.issue_id)
903 self._complete_issue_lifecycle_if_needed(result.issue_id)
904 else:
905 self.queue.mark_failed(result.issue_id)
906 else:
907 self.logger.error(f"{result.issue_id} failed: {result.error}")
908 self.queue.mark_failed(result.issue_id)
910 # Update timing
911 with self._state_lock:
912 self.state.timing[result.issue_id] = {
913 "total": result.duration,
914 }
916 # Clean up stage tracking after callback completes (ENH-262)
917 # Delay briefly so status reporter can show completion
918 self.worker_pool.remove_worker_stage(result.issue_id)
920 # Emit worker completion event for extensions (ENH-921)
921 if self._event_bus:
922 self._event_bus.emit(
923 {
924 "event": "parallel.worker_completed",
925 "ts": datetime.now(UTC).isoformat(),
926 "issue_id": result.issue_id,
927 "worker_name": result.worktree_path.name,
928 "status": "success" if result.success else "failure",
929 "duration_seconds": result.duration,
930 }
931 )
933 def _requeue_deferred_issues(self) -> None:
934 """Re-queue deferred issues that no longer have overlaps (ENH-143)."""
935 if not self._deferred_issues:
936 return
938 # Check each deferred issue for remaining overlaps
939 still_deferred = []
940 for issue in self._deferred_issues:
941 if self.overlap_detector:
942 overlap = self.overlap_detector.check_overlap(issue)
943 if overlap:
944 # Still has overlaps, keep deferred
945 still_deferred.append(issue)
946 else:
947 # No more overlaps, add back to queue
948 self.logger.info(f"Re-queuing {issue.issue_id} - no longer overlapping")
949 self.queue.add(issue)
951 self._deferred_issues = still_deferred
953 def _merge_sequential(self, result: WorkerResult) -> None:
954 """Merge a sequential (P0) result immediately.
956 Args:
957 result: Result to merge
958 """
959 # Handle closure for sequential issues
960 if result.should_close:
961 # Lazy import to avoid circular dependency
962 from little_loops.issue_lifecycle import close_issue
964 info = self._issue_info_by_id.get(result.issue_id)
965 if info and close_issue(
966 info,
967 self.br_config,
968 self.logger,
969 result.close_reason,
970 result.close_status,
971 interceptors=None,
972 ):
973 self.queue.mark_completed(result.issue_id)
974 else:
975 self.queue.mark_failed(result.issue_id)
976 return
978 self.merge_coordinator.queue_merge(result)
979 # Wait for this specific merge
980 self.merge_coordinator.wait_for_completion(timeout=60)
982 if result.issue_id in self.merge_coordinator.merged_ids:
983 self.queue.mark_completed(result.issue_id)
984 self._complete_issue_lifecycle_if_needed(result.issue_id)
985 else:
986 self.queue.mark_failed(result.issue_id)
988 def _wait_for_completion(self) -> None:
989 """Wait for all workers and merges to complete."""
990 self.logger.info("Waiting for workers to complete...")
992 # Calculate timeout
993 if self.parallel_config.orchestrator_timeout > 0:
994 timeout = self.parallel_config.orchestrator_timeout
995 else:
996 timeout = self.parallel_config.timeout_per_issue * self.parallel_config.max_workers
998 start = time.time()
999 while self.worker_pool.active_count > 0:
1000 if time.time() - start > timeout:
1001 self.logger.warning(f"Timeout waiting for workers after {timeout}s")
1002 self.worker_pool.terminate_all_processes()
1003 break
1004 time.sleep(1.0)
1006 # Wait for merges
1007 self.logger.info("Waiting for pending merges...")
1008 self.merge_coordinator.wait_for_completion(timeout=120)
1010 # Update queue with merge results and complete lifecycle
1011 for issue_id in self.merge_coordinator.merged_ids:
1012 self.queue.mark_completed(issue_id)
1013 self._complete_issue_lifecycle_if_needed(issue_id)
1015 for issue_id in self.merge_coordinator.failed_merges:
1016 self.queue.mark_failed(issue_id)
1018 def _report_results(self, start_time: float) -> None:
1019 """Report processing results.
1021 Args:
1022 start_time: When processing started
1023 """
1024 total_time = time.time() - start_time
1025 self._execution_duration = total_time
1027 self.logger.info("")
1028 self.logger.info("=" * 60)
1029 if self.wave_label:
1030 self.logger.info(f"{self.wave_label.upper()} PROCESSING COMPLETE")
1031 else:
1032 self.logger.info("PARALLEL ISSUE PROCESSING COMPLETE")
1033 self.logger.info("=" * 60)
1034 self.logger.info("")
1035 self.logger.timing(f"Total time: {format_duration(total_time)}")
1036 self.logger.info(f"Completed: {self.queue.completed_count}")
1037 self.logger.info(f"Failed: {self.queue.failed_count}")
1038 if self._interrupted_issues:
1039 self.logger.info(f"Interrupted: {len(self._interrupted_issues)}")
1041 with self._state_lock:
1042 timing_snapshot = dict(self.state.timing)
1043 corrections_snapshot = dict(self.state.corrections)
1045 if self.queue.completed_count > 0:
1046 total_issue_time = sum(t.get("total", 0) for t in timing_snapshot.values())
1047 if total_issue_time > 0:
1048 speedup = total_issue_time / total_time
1049 self.logger.info(f"Estimated speedup: {speedup:.2f}x")
1051 if self.queue.failed_ids:
1052 self.logger.info("")
1053 self.logger.warning("Failed issues:")
1054 for issue_id in self.queue.failed_ids:
1055 self.logger.warning(f" - {issue_id}")
1057 # Report interrupted issues separately (ENH-036)
1058 if self._interrupted_issues:
1059 self.logger.info("")
1060 self.logger.info(f"Interrupted: {len(self._interrupted_issues)} (can retry)")
1061 for issue_id in self._interrupted_issues:
1062 self.logger.info(f" - {issue_id}")
1064 # Report PR-ready branches when use_feature_branches=True (ENH-665)
1065 if self._pr_ready_branches:
1066 self.logger.info("")
1067 self.logger.info(f"PR-ready: {len(self._pr_ready_branches)} branch(es)")
1068 for issue_id, branch in self._pr_ready_branches.items():
1069 self.logger.info(f" - {issue_id}: {branch}")
1071 # Report correction statistics for quality tracking (ENH-010)
1072 if corrections_snapshot:
1073 total_corrected = len(corrections_snapshot)
1074 total_issues = self.queue.completed_count + self.queue.failed_count
1075 correction_rate = (total_corrected / total_issues * 100) if total_issues > 0 else 0
1076 self.logger.info("")
1077 self.logger.info(
1078 f"Auto-corrections: {total_corrected}/{total_issues} ({correction_rate:.1f}%)"
1079 )
1081 # Group corrections by category (ENH-010 fourth fix)
1082 from collections import Counter, defaultdict
1084 all_corrections: list[str] = []
1085 by_category: dict[str, int] = defaultdict(int)
1086 for corrections in corrections_snapshot.values():
1087 all_corrections.extend(corrections)
1088 for correction in corrections:
1089 # Extract category from [category] prefix if present
1090 if correction.startswith("[") and "]" in correction:
1091 category = correction[1 : correction.index("]")]
1092 by_category[category] += 1
1093 else:
1094 by_category["uncategorized"] += 1
1096 # Log corrections by type/category
1097 if by_category:
1098 self.logger.info("Corrections by type:")
1099 for category, count in sorted(by_category.items(), key=lambda x: -x[1]):
1100 self.logger.info(f" - {category}: {count}")
1102 # Log most common individual corrections
1103 if all_corrections:
1104 common = Counter(all_corrections).most_common(3)
1105 self.logger.info("Most common corrections:")
1106 for correction, count in common:
1107 # Truncate long correction descriptions
1108 display = correction[:60] + "..." if len(correction) > 60 else correction
1109 self.logger.info(f" - {display}: {count}")
1111 # Report stash pop warnings (local changes need manual recovery)
1112 stash_warnings = self.merge_coordinator.stash_pop_failures
1113 if stash_warnings:
1114 self.logger.info("")
1115 self.logger.warning("Stash recovery warnings (local changes need manual restoration):")
1116 for issue_id, message in stash_warnings.items():
1117 self.logger.warning(f" - {issue_id}: {message}")
1118 self.logger.warning("")
1119 self.logger.warning(
1120 "To recover: Run 'git stash list' to find your changes, "
1121 "then 'git stash pop' or 'git stash apply stash@{N}'"
1122 )
1124 def _complete_issue_lifecycle_if_needed(self, issue_id: str) -> bool:
1125 """Complete issue lifecycle if the issue file wasn't moved during merge.
1127 Args:
1128 issue_id: ID of the issue to complete
1130 Returns:
1131 True if lifecycle was completed (or already complete), False on error
1132 """
1133 info = self._issue_info_by_id.get(issue_id)
1134 if not info:
1135 self.logger.warning(f"No issue info found for {issue_id}")
1136 return False
1138 original_path = info.path
1139 completed_dir = self.br_config.get_completed_dir()
1140 completed_path = completed_dir / original_path.name
1142 # Check if already moved to completed
1143 if completed_path.exists():
1144 return True
1146 # Check if still in original location
1147 if not original_path.exists():
1148 return True
1150 # Issue file still in original location - complete lifecycle
1151 self.logger.info(f"Completing lifecycle for {issue_id} (merged but file not moved)")
1153 try:
1154 completed_dir.mkdir(parents=True, exist_ok=True)
1156 # Read original content
1157 content = original_path.read_text()
1159 # Add resolution section if not already present
1160 if "## Resolution" not in content:
1161 action = self.br_config.get_category_action(info.issue_type)
1162 resolution = f"""
1164---
1166## Resolution
1168- **Action**: {action}
1169- **Completed**: {datetime.now().strftime("%Y-%m-%d")}
1170- **Status**: Completed (parallel merge fallback)
1171- **Implementation**: Merged from parallel worker branch
1173### Changes Made
1174- See git history for implementation details
1176### Verification Results
1177- Work verification passed before merge
1179### Commits
1180- See `git log --oneline` for merge commit details
1181"""
1182 content += resolution
1184 # Use git mv if possible (before writing content to avoid "destination exists")
1185 result = self._git_lock.run(
1186 ["mv", str(original_path), str(completed_path)],
1187 cwd=self.repo_path,
1188 )
1190 if result.returncode != 0:
1191 # git mv failed (destination may exist or other error)
1192 self.logger.warning(f"git mv failed for {issue_id}: {result.stderr}")
1193 # Write content to destination (may overwrite existing)
1194 completed_path.write_text(content)
1195 # Remove source if it still exists
1196 original_path.unlink(missing_ok=True)
1197 else:
1198 # git mv succeeded, write updated content
1199 completed_path.write_text(content)
1200 append_session_log_entry(completed_path, "ll-parallel")
1202 # Stage and commit
1203 self._git_lock.run(
1204 ["add", "-A"],
1205 cwd=self.repo_path,
1206 )
1208 action = self.br_config.get_category_action(info.issue_type)
1209 commit_msg = f"""{action}({info.issue_type}): complete {issue_id} lifecycle
1211Parallel merge fallback - issue file moved to completed.
1213Issue: {issue_id}
1214Type: {info.issue_type}
1215Title: {info.title}
1216"""
1217 commit_result = self._git_lock.run(
1218 ["commit", "-m", commit_msg],
1219 cwd=self.repo_path,
1220 )
1222 if commit_result.returncode != 0:
1223 if "nothing to commit" not in commit_result.stdout.lower():
1224 self.logger.warning(f"git commit failed: {commit_result.stderr}")
1225 else:
1226 commit_hash_match = re.search(r"\[[\w-]+\s+([a-f0-9]+)\]", commit_result.stdout)
1227 if commit_hash_match:
1228 self.logger.success(
1229 f"Completed lifecycle for {issue_id}: {commit_hash_match.group(1)}"
1230 )
1231 else:
1232 self.logger.success(f"Completed lifecycle for {issue_id}")
1234 return True
1236 except Exception as e:
1237 self.logger.error(f"Failed to complete lifecycle for {issue_id}: {e}")
1238 return False
1240 def _cleanup(self) -> None:
1241 """Clean up resources."""
1242 self.logger.info("Cleaning up...")
1244 # Save final state (force=True bypasses throttle to ensure shutdown state is persisted)
1245 self._save_state(force=True)
1247 # Shutdown components
1248 self.worker_pool.shutdown(wait=True)
1249 self.merge_coordinator.shutdown(wait=True, timeout=30)
1251 # Clean up worktrees if not interrupted
1252 if not self._shutdown_requested:
1253 self.worker_pool.cleanup_all_worktrees()