Coverage for little_loops / parallel / orchestrator.py: 8%
705 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -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 parse_frontmatter, 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_branch, _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, dry_run: bool = False) -> 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.
244 Args:
245 dry_run: If True, log what would be removed but make no changes.
246 """
247 worktree_base = self.repo_path / self.parallel_config.worktree_base
248 if not worktree_base.exists():
249 return
251 # Get list of worktree directories, skipping those owned by live processes (BUG-579)
252 orphaned = []
253 for item in worktree_base.iterdir():
254 if item.is_dir() and _is_ll_worktree(item.name):
255 # Check for a .ll-session-<pid> marker left by an active orchestrator
256 owned_by_live = False
257 for marker in item.glob(".ll-session-*"):
258 try:
259 pid = int(marker.name.split("-")[-1])
260 os.kill(pid, 0) # Signal 0: check if process exists
261 owned_by_live = True
262 break
263 except (ProcessLookupError, ValueError):
264 pass
265 except PermissionError:
266 owned_by_live = True # Process exists, no permission to signal
267 break
268 if owned_by_live:
269 self.logger.info(f"Skipping {item.name}: owned by running process")
270 continue
271 orphaned.append(item)
273 if orphaned:
274 self.logger.info(f"Cleaning up {len(orphaned)} orphaned worktree(s) from previous run")
276 for worktree_path in orphaned:
277 try:
278 # Resolve branch name BEFORE removing the worktree (BUG-2324)
279 branch_result = subprocess.run(
280 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
281 cwd=worktree_path,
282 capture_output=True,
283 text=True,
284 )
285 branch_name = (
286 branch_result.stdout.strip() if branch_result.returncode == 0 else None
287 )
289 if dry_run:
290 self.logger.info(
291 f"[dry-run] Would remove worktree {worktree_path.name}"
292 + (
293 f" and branch {branch_name}"
294 if branch_name and _is_ll_branch(branch_name)
295 else ""
296 )
297 )
298 continue
300 self._git_lock.run(
301 ["worktree", "unlock", str(worktree_path)],
302 cwd=self.repo_path,
303 timeout=10,
304 )
305 # Try git worktree remove first
306 self._git_lock.run(
307 ["worktree", "remove", "--force", str(worktree_path)],
308 cwd=self.repo_path,
309 timeout=30,
310 )
312 # If git worktree remove failed, force delete the directory
313 if worktree_path.exists():
314 shutil.rmtree(worktree_path, ignore_errors=True)
316 # Delete branch only for ll-managed shapes (BUG-2324: safe guard replaces
317 # the old parallel/-only guard to also cover loop worktree branches)
318 if branch_name and _is_ll_branch(branch_name):
319 self._git_lock.run(
320 ["branch", "-D", branch_name],
321 cwd=self.repo_path,
322 timeout=10,
323 )
324 except Exception as e:
325 self.logger.warning(f"Failed to clean up {worktree_path.name}: {e}")
327 if not dry_run:
328 # Also prune git worktree references
329 self._git_lock.run(
330 ["worktree", "prune"],
331 cwd=self.repo_path,
332 timeout=30,
333 )
335 self._prune_ghost_worktree_refs()
337 def _prune_ghost_worktree_refs(self) -> None:
338 """Prune git worktree metadata entries whose on-disk path no longer exists.
340 Handles the SIGKILL race where a worker directory was deleted before
341 git worktree prune ran, leaving .git/worktrees/<name>/ intact. The
342 next git worktree add for the same path would then fail with "already exists".
343 """
344 try:
345 result = self._git_lock.run(
346 ["worktree", "list", "--porcelain"],
347 cwd=self.repo_path,
348 timeout=30,
349 )
350 except Exception as e:
351 self.logger.warning(f"Failed to list worktrees for ghost ref scan: {e}")
352 return
354 ghost_names: list[str] = []
355 current: dict[str, str] = {}
356 for line in result.stdout.splitlines():
357 if not line:
358 if current:
359 path_str = current.get("worktree", "")
360 name = Path(path_str).name
361 if _is_ll_worktree(name) and path_str and not Path(path_str).exists():
362 ghost_names.append(name)
363 current = {}
364 continue
365 key, _, value = line.partition(" ")
366 current[key] = value
367 if current:
368 path_str = current.get("worktree", "")
369 name = Path(path_str).name
370 if _is_ll_worktree(name) and path_str and not Path(path_str).exists():
371 ghost_names.append(name)
373 if not ghost_names:
374 return
376 for name in ghost_names:
377 self.logger.info(f"Pruned ghost ref: {name}")
379 try:
380 self._git_lock.run(
381 ["worktree", "prune"],
382 cwd=self.repo_path,
383 timeout=30,
384 )
385 except Exception as e:
386 self.logger.warning(f"Failed to prune ghost worktree refs: {e}")
388 def _inspect_worktree(self, worktree_path: Path) -> PendingWorktreeInfo | None:
389 """Inspect a worktree to determine its status.
391 Args:
392 worktree_path: Path to the worktree directory
394 Returns:
395 PendingWorktreeInfo if inspection succeeded, None if failed
396 """
397 try:
398 # Read actual branch name from worktree via rev-parse
399 branch_result = subprocess.run(
400 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
401 cwd=worktree_path,
402 capture_output=True,
403 text=True,
404 )
405 if branch_result.returncode == 0:
406 branch_name = branch_result.stdout.strip()
407 else:
408 # Fall back to string derivation if rev-parse fails
409 branch_name = worktree_path.name.replace("worker-", "parallel/")
411 # Extract issue ID (e.g., bug-045 -> BUG-045)
412 # Pattern: worker-<issue-id>-<timestamp>
413 match = re.match(r"worker-([a-z]+-\d+)-\d{8}-\d{6}", worktree_path.name)
414 issue_id = match.group(1).upper() if match else worktree_path.name
416 # Check commits ahead of main
417 result = self._git_lock.run(
418 ["rev-list", "--count", f"{self.parallel_config.base_branch}..{branch_name}"],
419 cwd=self.repo_path,
420 timeout=10,
421 )
422 commits_ahead = int(result.stdout.strip()) if result.returncode == 0 else 0
424 # Check for uncommitted changes in worktree
425 result = self._git_lock.run(
426 ["status", "--porcelain"],
427 cwd=worktree_path,
428 timeout=10,
429 )
430 changed_files = []
431 has_uncommitted = False
432 if result.returncode == 0 and result.stdout.strip():
433 has_uncommitted = True
434 changed_files = [line[3:] for line in result.stdout.strip().split("\n") if line]
436 return PendingWorktreeInfo(
437 worktree_path=worktree_path,
438 branch_name=branch_name,
439 issue_id=issue_id,
440 commits_ahead=commits_ahead,
441 has_uncommitted_changes=has_uncommitted,
442 changed_files=changed_files,
443 )
444 except Exception as e:
445 self.logger.warning(f"Failed to inspect worktree {worktree_path.name}: {e}")
446 return None
448 def _check_pending_worktrees(self) -> list[PendingWorktreeInfo]:
449 """Check for pending worktrees from previous runs and report status.
451 Returns:
452 List of pending worktree information
453 """
454 worktree_base = self.repo_path / self.parallel_config.worktree_base
455 if not worktree_base.exists():
456 return []
458 # Find all worker directories
459 worktrees = [
460 item for item in worktree_base.iterdir() if item.is_dir() and _is_ll_worktree(item.name)
461 ]
463 if not worktrees:
464 return []
466 self.logger.info("Checking for pending work from previous runs...")
468 # Inspect each worktree
469 pending_info: list[PendingWorktreeInfo] = []
470 for worktree_path in worktrees:
471 info = self._inspect_worktree(worktree_path)
472 if info:
473 pending_info.append(info)
475 # Report findings
476 with_work = [p for p in pending_info if p.has_pending_work]
477 if with_work:
478 self.logger.warning(f"Found {len(with_work)} worktree(s) with pending work:")
479 for info in with_work:
480 status_parts = []
481 if info.commits_ahead > 0:
482 status_parts.append(f"{info.commits_ahead} commit(s) ahead of main")
483 if info.has_uncommitted_changes:
484 status_parts.append(f"{len(info.changed_files)} uncommitted file(s)")
485 status = ", ".join(status_parts)
486 self.logger.warning(f" - {info.worktree_path.name}: {info.issue_id} ({status})")
488 self.logger.info("")
489 self.logger.info("Options:")
490 self.logger.info(" --merge-pending Attempt to merge pending work before continuing")
491 self.logger.info(" --clean-start Remove all worktrees and start fresh")
492 self.logger.info(
493 " --ignore-pending Continue without action (worktrees will be cleaned up)"
494 )
495 elif pending_info:
496 self.logger.info(f"Found {len(pending_info)} orphaned worktree(s) with no pending work")
498 return pending_info
500 def _merge_pending_worktrees(self, pending: list[PendingWorktreeInfo]) -> None:
501 """Attempt to merge pending worktrees from previous runs.
503 Args:
504 pending: List of pending worktree information
505 """
506 with_work = [p for p in pending if p.has_pending_work]
507 if not with_work:
508 return
510 self.logger.info(f"Attempting to merge {len(with_work)} pending worktree(s)...")
512 for info in with_work:
513 try:
514 # If there are uncommitted changes, commit them first
515 if info.has_uncommitted_changes:
516 self.logger.info(f" Committing uncommitted changes in {info.issue_id}...")
517 self._git_lock.run(
518 ["add", "-A"],
519 cwd=info.worktree_path,
520 timeout=30,
521 )
522 self._git_lock.run(
523 [
524 "commit",
525 "-m",
526 f"WIP: Auto-commit from interrupted session for {info.issue_id}",
527 ],
528 cwd=info.worktree_path,
529 timeout=30,
530 )
532 # Attempt merge
533 self.logger.info(f" Merging {info.issue_id} ({info.branch_name})...")
534 result = self._git_lock.run(
535 [
536 "merge",
537 "--no-ff",
538 info.branch_name,
539 "-m",
540 f"Merge pending work for {info.issue_id}",
541 ],
542 cwd=self.repo_path,
543 timeout=60,
544 )
546 if result.returncode == 0:
547 self.logger.success(f" Successfully merged {info.issue_id}")
548 # Clean up the worktree after successful merge
549 self._git_lock.run(
550 ["worktree", "remove", "--force", str(info.worktree_path)],
551 cwd=self.repo_path,
552 timeout=30,
553 )
554 self._git_lock.run(
555 ["branch", "-D", info.branch_name],
556 cwd=self.repo_path,
557 timeout=10,
558 )
559 else:
560 self.logger.warning(f" Failed to merge {info.issue_id}: {result.stderr}")
561 # Abort the merge if it failed
562 self._git_lock.run(
563 ["merge", "--abort"],
564 cwd=self.repo_path,
565 timeout=10,
566 )
568 except Exception as e:
569 self.logger.warning(f" Error merging {info.issue_id}: {e}")
571 def _signal_handler(self, signum: int, frame: object) -> None:
572 """Handle shutdown signals gracefully."""
573 self._shutdown_requested = True
574 # Propagate to worker pool for interrupted worker detection (ENH-036)
575 self.worker_pool.set_shutdown_requested(True)
576 self.logger.warning(f"Received signal {signum}, shutting down gracefully...")
578 def _load_state(self) -> None:
579 """Load state from file for resume capability."""
580 if self.parallel_config.clean_start:
581 self.state.started_at = datetime.now().isoformat()
582 return
583 state_file = self.repo_path / self.parallel_config.state_file
584 if not state_file.exists():
585 self.state.started_at = datetime.now().isoformat()
586 return
588 try:
589 data = json.loads(state_file.read_text())
590 self.state = OrchestratorState.from_dict(data)
592 # Restore queue state
593 self.queue.load_completed(self.state.completed_issues)
594 self.queue.load_failed(self.state.failed_issues.keys())
596 self.logger.info(
597 f"Resumed from previous state: "
598 f"{len(self.state.completed_issues)} completed, "
599 f"{len(self.state.failed_issues)} failed"
600 )
601 except Exception as e:
602 self.logger.warning(f"Could not load state: {e}")
603 self.state.started_at = datetime.now().isoformat()
605 def _save_state(self, force: bool = False) -> None:
606 """Save current state to file using an atomic write.
608 Writes are throttled to at most once every 5 seconds to reduce filesystem I/O
609 during high-frequency loop ticks (e.g., merge-waiting phase). Pass force=True
610 to bypass the throttle, e.g., on shutdown.
611 """
612 now = time.time()
613 if not force and now - self._last_save_time < 5.0:
614 return
615 self._last_save_time = now
616 with self._state_lock:
617 self.state.last_checkpoint = datetime.now().isoformat()
618 self.state.completed_issues = self.queue.completed_ids
619 self.state.failed_issues = {
620 issue_id: self._worker_errors.get(issue_id, "Failed")
621 for issue_id in self.queue.failed_ids
622 }
623 self.state.in_progress_issues = self.queue.in_progress_ids
625 state_file = self.repo_path / self.parallel_config.state_file
626 data = json.dumps(self.state.to_dict(), indent=2)
627 tmp_fd, tmp_path = tempfile.mkstemp(dir=state_file.parent, suffix=".tmp")
628 try:
629 with os.fdopen(tmp_fd, "w") as f:
630 f.write(data)
631 os.replace(tmp_path, state_file)
632 except Exception:
633 os.unlink(tmp_path)
634 raise
636 def _cleanup_state(self) -> None:
637 """Remove state file on successful completion."""
638 state_file = self.repo_path / self.parallel_config.state_file
639 if state_file.exists():
640 state_file.unlink()
642 def _dry_run(self) -> int:
643 """Preview what would be processed without executing.
645 Returns:
646 Exit code (always 0 for dry run)
647 """
648 issues = self._scan_issues()
650 self.logger.info("=" * 60)
651 self.logger.info("DRY RUN - No changes will be made")
652 self.logger.info("=" * 60)
653 self.logger.info("")
655 if not issues:
656 self.logger.info("No issues found matching criteria")
657 return 0
659 self.logger.info(f"Found {len(issues)} issues to process:")
660 self.logger.info("")
662 # Group by priority
663 by_priority: dict[str, list[IssueInfo]] = {}
664 for issue in issues:
665 by_priority.setdefault(issue.priority, []).append(issue)
667 for priority in IssuePriorityQueue.DEFAULT_PRIORITIES:
668 if priority not in by_priority:
669 continue
671 priority_issues = by_priority[priority]
672 self.logger.info(f" {priority} ({len(priority_issues)} issues):")
673 for issue in priority_issues:
674 mode = (
675 "sequential"
676 if priority == "P0" and self.parallel_config.p0_sequential
677 else "parallel"
678 )
679 self.logger.info(f" - {issue.issue_id}: {issue.title} [{mode}]")
681 self.logger.info("")
682 self.logger.info("Configuration:")
683 self.logger.info(f" Workers: {self.parallel_config.max_workers}")
684 self.logger.info(f" P0 Sequential: {self.parallel_config.p0_sequential}")
685 self.logger.info(f" Max Issues: {self.parallel_config.max_issues or 'unlimited'}")
686 self.logger.info(f" Command Prefix: {self.parallel_config.command_prefix}")
688 return 0
690 def _maybe_report_status(self) -> None:
691 """Report status if enough time has elapsed since last report.
693 Reports every 5 seconds during active processing for progress visibility (ENH-262).
694 Suppresses duplicate lines when nothing has changed.
695 """
696 now = time.time()
697 # Report every 5 seconds
698 if now - self._last_status_time < 5.0:
699 return
701 self._last_status_time = now
703 # Build status line
704 parts = []
706 # Add wave label if present
707 if self.wave_label:
708 parts.append(f"{self.wave_label}")
710 # Get queue counts
711 in_progress = len(self.queue.in_progress_ids)
712 completed = self.queue.completed_count
713 failed = self.queue.failed_count
714 pending_merge = self.merge_coordinator.pending_count
716 parts.append(f"Active: {in_progress}")
717 parts.append(f"Done: {completed}")
718 if failed > 0:
719 parts.append(f"Failed: {failed}")
720 if pending_merge > 0:
721 parts.append(f"Merging: {pending_merge}")
723 # Build status line
724 status = " | ".join(parts)
726 # Get active worker stages
727 active_stages = self.worker_pool.get_active_stages()
729 # Add worker details if any are active
730 if active_stages:
731 # Group by stage
732 by_stage: dict[str, list[str]] = {}
733 for issue_id, worker_stage in active_stages.items():
734 stage_name = worker_stage.value.title()
735 by_stage.setdefault(stage_name, []).append(issue_id)
737 stage_parts = []
738 for stage_name in ["Validating", "Implementing", "Verifying", "Merging"]:
739 if stage_name in by_stage:
740 issue_ids = ", ".join(by_stage[stage_name])
741 stage_parts.append(f"{stage_name}: [{issue_ids}]")
743 if stage_parts:
744 status += " | " + " | ".join(stage_parts)
746 # Skip if nothing changed since last report
747 if status == self._last_status_line:
748 return
749 self._last_status_line = status
751 # Log with gray color to distinguish from normal logs
752 self.logger.debug(status)
754 def _execute(self) -> int:
755 """Execute parallel issue processing.
757 Returns:
758 Exit code (0 = success, 1 = failure)
759 """
760 start_time = time.time()
762 # Scan and queue issues
763 issues = self._scan_issues()
764 if not issues:
765 self.logger.info("No issues to process")
766 return 0
768 # Store issue info for lifecycle completion after merge
769 for issue in issues:
770 self._issue_info_by_id[issue.issue_id] = issue
772 added = self.queue.add_many(issues)
773 self.logger.info(f"Queued {added} issues for processing")
775 # Start components
776 self.worker_pool.start()
777 self.merge_coordinator.start()
779 # Process issues
780 issues_processed = 0
781 max_issues = self.parallel_config.max_issues or float("inf")
783 while not self._shutdown_requested:
784 # Check if done
785 if self.queue.empty() and self.worker_pool.active_count == 0:
786 # Wait for pending merges
787 if self.merge_coordinator.pending_count == 0:
788 break
790 # Check max issues limit
791 if issues_processed >= max_issues:
792 self.logger.info(f"Reached max issues limit ({max_issues})")
793 break
795 # Get next issue if workers available
796 if self.worker_pool.active_count < self.parallel_config.max_workers:
797 queued = self.queue.get(block=False)
798 if queued:
799 issue = queued.issue_info
801 # P0 sequential processing
802 if issue.priority == "P0" and self.parallel_config.p0_sequential:
803 self._process_sequential(issue)
804 else:
805 self._process_parallel(issue)
807 issues_processed += 1
809 # Save state periodically
810 self._save_state()
812 # Report status periodically for progress visibility (ENH-262)
813 self._maybe_report_status()
815 # Small sleep to prevent busy loop
816 time.sleep(0.1)
818 # Wait for completion
819 self._wait_for_completion()
821 # Report results
822 self._report_results(start_time)
824 # Cleanup state on success
825 if not self._shutdown_requested and self.queue.failed_count == 0:
826 self._cleanup_state()
828 return 0 if self.queue.failed_count == 0 else 1
830 def _scan_issues(self) -> list[IssueInfo]:
831 """Scan for issues matching criteria.
833 Returns:
834 List of issues sorted by priority
835 """
836 # Combine skip_ids from state and config
837 skip_ids = set(self.state.completed_issues) | set(self.state.failed_issues.keys())
838 if self.parallel_config.skip_ids:
839 skip_ids |= self.parallel_config.skip_ids
841 issues = IssuePriorityQueue.scan_issues(
842 self.br_config,
843 priority_filter=list(self.parallel_config.priority_filter),
844 skip_ids=skip_ids,
845 only_ids=self.parallel_config.only_ids,
846 type_prefixes=self.parallel_config.type_prefixes,
847 label_filter=self.parallel_config.label_filter,
848 )
850 # Apply max issues limit
851 if self.parallel_config.max_issues > 0:
852 issues = issues[: self.parallel_config.max_issues]
854 return issues
856 def _process_sequential(self, issue: IssueInfo) -> None:
857 """Process an issue sequentially (blocking).
859 Args:
860 issue: Issue to process
861 """
862 self.logger.info(f"Processing {issue.issue_id} sequentially (P0)")
864 # Wait for any parallel work to finish
865 while self.worker_pool.active_count > 0:
866 time.sleep(0.5)
868 # Process in main repo (no worktree isolation needed)
869 # Note: No callback here - _merge_sequential handles the result explicitly
870 # to avoid double-processing (callback would also queue merge/close)
871 future = self.worker_pool.submit(issue)
873 # Wait for completion
874 try:
875 result = future.result(timeout=self.parallel_config.timeout_per_issue)
876 if result.success:
877 # Merge immediately for P0
878 self._merge_sequential(result)
879 except Exception as e:
880 self.logger.error(f"Sequential processing failed: {e}")
881 self.queue.mark_failed(issue.issue_id)
883 def _process_parallel(self, issue: IssueInfo) -> None:
884 """Process an issue in parallel (non-blocking).
886 Args:
887 issue: Issue to process
888 """
889 # Check for overlaps if enabled (ENH-143)
890 if self.overlap_detector:
891 overlap = self.overlap_detector.check_overlap(issue)
892 if overlap:
893 if self.parallel_config.serialize_overlapping:
894 self.logger.warning(
895 f"Deferring {issue.issue_id} - overlaps with {overlap.overlapping_issues}"
896 )
897 # Track for re-check when active issues complete
898 self._deferred_issues.append(issue)
899 return
900 else:
901 self.logger.warning(
902 f"Warning: {issue.issue_id} may conflict with {overlap.overlapping_issues}"
903 )
905 # Register as active before dispatch
906 self.overlap_detector.register_issue(issue)
908 self.logger.info(f"Dispatching {issue.issue_id} to worker pool")
909 self.worker_pool.submit(issue, self._on_worker_complete)
911 def _on_worker_complete(self, result: WorkerResult) -> None:
912 """Callback when a worker completes.
914 Args:
915 result: Result from the worker
916 """
917 # Unregister from overlap tracking (ENH-143)
918 if self.overlap_detector:
919 self.overlap_detector.unregister_issue(result.issue_id)
920 # Re-queue deferred issues that were waiting on this one
921 self._requeue_deferred_issues()
923 # Handle interrupted workers (not counted as failed) - ENH-036
924 if result.interrupted:
925 self.logger.info(f"{result.issue_id} was interrupted during shutdown (can retry)")
926 self._interrupted_issues.append(result.issue_id)
927 # Don't mark as failed - they can be retried on next run
928 return
930 # Handle issue closure (no merge needed)
931 if result.should_close:
932 # Lazy import to avoid circular dependency
933 from little_loops.issue_lifecycle import close_issue
935 self.logger.info(f"{result.issue_id} should be closed: {result.close_status}")
936 info = self._issue_info_by_id.get(result.issue_id)
937 if info:
938 # TODO(ENH-1686): parallel-path close events not yet live-written
939 if close_issue(
940 info,
941 self.br_config,
942 self.logger,
943 result.close_reason,
944 result.close_status,
945 interceptors=None,
946 ):
947 self.queue.mark_completed(result.issue_id)
948 else:
949 self._worker_errors[result.issue_id] = (
950 f"Close failed: {result.close_reason or 'close error'}"
951 )
952 self.queue.mark_failed(result.issue_id)
953 else:
954 self.logger.warning(f"No issue info found for {result.issue_id}")
955 self._worker_errors[result.issue_id] = "Close failed: no issue info"
956 self.queue.mark_failed(result.issue_id)
957 elif result.success:
958 self.logger.success(
959 f"{result.issue_id} completed in {format_duration(result.duration)}"
960 )
961 if result.was_corrected:
962 self.logger.info(f"{result.issue_id} was auto-corrected during validation")
963 # Log and store corrections for pattern analysis (ENH-010)
964 for correction in result.corrections:
965 self.logger.info(f" Correction: {correction}")
966 if result.corrections:
967 with self._state_lock:
968 self.state.corrections[result.issue_id] = result.corrections
969 if self.parallel_config.use_feature_branches:
970 # Feature branch mode: skip auto-merge, branch stays alive (ENH-665, BUG-2172)
971 # Hold issue at in_progress until PR is merged; ll-sync reconcile promotes to done (ENH-2182)
972 self.logger.info(f"{result.issue_id}: feature branch ready — {result.branch_name}")
973 self.queue.mark_completed(result.issue_id)
974 self._complete_issue_lifecycle_if_needed(
975 result.issue_id, terminal_status="in_progress"
976 )
977 branch_state: dict[str, Any] = {
978 "branch_name": result.branch_name,
979 "pushed": False,
980 "pr_url": None,
981 }
982 if self.parallel_config.push_feature_branches:
983 push_result = subprocess.run(
984 [
985 "git",
986 "push",
987 "--force-with-lease",
988 self.parallel_config.remote_name,
989 result.branch_name,
990 ],
991 cwd=self.repo_path,
992 capture_output=True,
993 text=True,
994 timeout=60,
995 )
996 if push_result.returncode == 0:
997 branch_state["pushed"] = True
998 self.logger.info(
999 f"{result.issue_id}: pushed {result.branch_name}"
1000 f" to {self.parallel_config.remote_name}"
1001 )
1002 if self.parallel_config.open_pr_for_feature_branches:
1003 self._open_pr_for_branch(
1004 result.issue_id, result.branch_name, branch_state
1005 )
1006 else:
1007 self.logger.warning(
1008 f"{result.issue_id}: git push failed: {push_result.stderr.strip()}"
1009 )
1010 # Write branch name (and optional PR URL) back to issue frontmatter (ENH-2175)
1011 info = self._issue_info_by_id.get(result.issue_id)
1012 if info and info.path.exists():
1013 try:
1014 content = info.path.read_text()
1015 fm = parse_frontmatter(content)
1016 updates: dict[str, Any] = {"branch": result.branch_name}
1017 if branch_state.get("pr_url") and not fm.get("pr_url"):
1018 updates["pr_url"] = branch_state["pr_url"]
1019 content = update_frontmatter(content, updates)
1020 info.path.write_text(content)
1021 self._git_lock.run(["add", "-A"], cwd=self.repo_path)
1022 commit_result = self._git_lock.run(
1023 [
1024 "commit",
1025 "-m",
1026 f"{result.issue_id}: record feature branch in frontmatter",
1027 ],
1028 cwd=self.repo_path,
1029 )
1030 if (
1031 commit_result.returncode != 0
1032 and "nothing to commit" not in commit_result.stdout.lower()
1033 ):
1034 self.logger.warning(
1035 f"{result.issue_id}: branch frontmatter commit failed:"
1036 f" {commit_result.stderr}"
1037 )
1038 except Exception as exc:
1039 self.logger.warning(
1040 f"{result.issue_id}: failed to record branch in frontmatter: {exc}"
1041 )
1042 with self._state_lock:
1043 self._pr_ready_branches[result.issue_id] = branch_state
1044 else:
1045 self.merge_coordinator.queue_merge(result)
1046 # Wait for merge to complete before returning from callback.
1047 # This prevents dispatch of next worker while merge is in progress,
1048 # avoiding race conditions between worktree creation and merge ops.
1049 # (BUG-140: Race condition between worktree creation and merge)
1050 self.merge_coordinator.wait_for_completion(timeout=120)
1051 if result.issue_id in self.merge_coordinator.merged_ids:
1052 self.queue.mark_completed(result.issue_id)
1053 self._complete_issue_lifecycle_if_needed(result.issue_id)
1054 else:
1055 self._worker_errors[result.issue_id] = (
1056 f"Merge failed: {result.error or 'merge error'}"
1057 )
1058 self.queue.mark_failed(result.issue_id)
1059 else:
1060 self.logger.error(f"{result.issue_id} failed: {result.error}")
1061 self._worker_errors[result.issue_id] = result.error or "Failed"
1062 self.queue.mark_failed(result.issue_id)
1064 # Update timing
1065 with self._state_lock:
1066 self.state.timing[result.issue_id] = {
1067 "total": result.duration,
1068 }
1070 # Clean up stage tracking after callback completes (ENH-262)
1071 # Delay briefly so status reporter can show completion
1072 self.worker_pool.remove_worker_stage(result.issue_id)
1074 # Emit worker completion event for extensions (ENH-921)
1075 if self._event_bus:
1076 self._event_bus.emit(
1077 {
1078 "event": "parallel.worker_completed",
1079 "ts": datetime.now(UTC).isoformat(),
1080 "issue_id": result.issue_id,
1081 "worker_name": result.worktree_path.name,
1082 "status": "success" if result.success else "failure",
1083 "duration_seconds": result.duration,
1084 }
1085 )
1087 def _requeue_deferred_issues(self) -> None:
1088 """Re-queue deferred issues that no longer have overlaps (ENH-143)."""
1089 if not self._deferred_issues:
1090 return
1092 # Check each deferred issue for remaining overlaps
1093 still_deferred = []
1094 for issue in self._deferred_issues:
1095 if self.overlap_detector:
1096 overlap = self.overlap_detector.check_overlap(issue)
1097 if overlap:
1098 # Still has overlaps, keep deferred
1099 still_deferred.append(issue)
1100 else:
1101 # No more overlaps, add back to queue
1102 self.logger.info(f"Re-queuing {issue.issue_id} - no longer overlapping")
1103 self.queue.add(issue)
1105 self._deferred_issues = still_deferred
1107 def _open_pr_for_branch(
1108 self,
1109 issue_id: str,
1110 branch_name: str,
1111 branch_state: dict[str, Any],
1112 ) -> None:
1113 """Open a draft PR for a pushed feature branch using the gh CLI.
1115 Mutates branch_state in place to record pr_url on success.
1116 Degrades gracefully if gh is missing or unauthenticated.
1117 """
1118 try:
1119 auth_result = subprocess.run(
1120 ["gh", "auth", "status"],
1121 capture_output=True,
1122 text=True,
1123 timeout=30,
1124 )
1125 if auth_result.returncode != 0:
1126 self.logger.warning(f"{issue_id}: gh not authenticated, skipping PR creation")
1127 return
1128 issue_info = self._issue_info_by_id.get(issue_id)
1129 pr_title = issue_info.title if issue_info else issue_id
1130 pr_result = subprocess.run(
1131 [
1132 "gh",
1133 "pr",
1134 "create",
1135 "--title",
1136 pr_title,
1137 "--body",
1138 f"Closes {issue_id}",
1139 "--base",
1140 self.parallel_config.base_branch,
1141 "--draft",
1142 "--head",
1143 branch_name,
1144 ],
1145 cwd=self.repo_path,
1146 capture_output=True,
1147 text=True,
1148 timeout=60,
1149 )
1150 if pr_result.returncode == 0:
1151 branch_state["pr_url"] = pr_result.stdout.strip()
1152 self.logger.info(f"{issue_id}: PR opened: {branch_state['pr_url']}")
1153 else:
1154 self.logger.warning(f"{issue_id}: gh pr create failed: {pr_result.stderr.strip()}")
1155 except FileNotFoundError:
1156 self.logger.warning(f"{issue_id}: gh CLI not found, skipping PR creation")
1157 except subprocess.TimeoutExpired:
1158 self.logger.warning(f"{issue_id}: gh pr create timed out")
1160 def _merge_sequential(self, result: WorkerResult) -> None:
1161 """Merge a sequential (P0) result immediately.
1163 Args:
1164 result: Result to merge
1165 """
1166 # Handle closure for sequential issues
1167 if result.should_close:
1168 # Lazy import to avoid circular dependency
1169 from little_loops.issue_lifecycle import close_issue
1171 info = self._issue_info_by_id.get(result.issue_id)
1172 # TODO(ENH-1686): parallel-path close events not yet live-written
1173 if info and close_issue(
1174 info,
1175 self.br_config,
1176 self.logger,
1177 result.close_reason,
1178 result.close_status,
1179 interceptors=None,
1180 ):
1181 self.queue.mark_completed(result.issue_id)
1182 else:
1183 self._worker_errors[result.issue_id] = (
1184 f"Close failed: {result.close_reason or 'close error'}"
1185 )
1186 self.queue.mark_failed(result.issue_id)
1187 return
1189 self.merge_coordinator.queue_merge(result)
1190 # Wait for this specific merge
1191 self.merge_coordinator.wait_for_completion(timeout=60)
1193 if result.issue_id in self.merge_coordinator.merged_ids:
1194 self.queue.mark_completed(result.issue_id)
1195 self._complete_issue_lifecycle_if_needed(result.issue_id)
1196 else:
1197 self._worker_errors[result.issue_id] = f"Merge failed: {result.error or 'merge error'}"
1198 self.queue.mark_failed(result.issue_id)
1200 def _wait_for_completion(self) -> None:
1201 """Wait for all workers and merges to complete."""
1202 self.logger.info("Waiting for workers to complete...")
1204 # Calculate timeout
1205 if self.parallel_config.orchestrator_timeout > 0:
1206 timeout = self.parallel_config.orchestrator_timeout
1207 else:
1208 timeout = self.parallel_config.timeout_per_issue * self.parallel_config.max_workers
1210 start = time.time()
1211 while self.worker_pool.active_count > 0:
1212 if time.time() - start > timeout:
1213 self.logger.warning(f"Timeout waiting for workers after {timeout}s")
1214 self.worker_pool.terminate_all_processes()
1215 break
1216 time.sleep(1.0)
1218 # Wait for merges
1219 self.logger.info("Waiting for pending merges...")
1220 self.merge_coordinator.wait_for_completion(timeout=120)
1222 # Update queue with merge results and complete lifecycle
1223 for issue_id in self.merge_coordinator.merged_ids:
1224 self.queue.mark_completed(issue_id)
1225 self._complete_issue_lifecycle_if_needed(issue_id)
1227 for issue_id, reason in self.merge_coordinator.failed_merges.items():
1228 self._worker_errors[issue_id] = reason or "Merge failed"
1229 self.queue.mark_failed(issue_id)
1231 def _report_results(self, start_time: float) -> None:
1232 """Report processing results.
1234 Args:
1235 start_time: When processing started
1236 """
1237 total_time = time.time() - start_time
1238 self._execution_duration = total_time
1240 self.logger.info("")
1241 self.logger.info("=" * 60)
1242 if self.wave_label:
1243 self.logger.info(f"{self.wave_label.upper()} PROCESSING COMPLETE")
1244 else:
1245 self.logger.info("PARALLEL ISSUE PROCESSING COMPLETE")
1246 self.logger.info("=" * 60)
1247 self.logger.info("")
1248 self.logger.timing(f"Total time: {format_duration(total_time)}")
1249 self.logger.info(f"Completed: {self.queue.completed_count}")
1250 self.logger.info(f"Failed: {self.queue.failed_count}")
1251 if self._interrupted_issues:
1252 self.logger.info(f"Interrupted: {len(self._interrupted_issues)}")
1254 with self._state_lock:
1255 timing_snapshot = dict(self.state.timing)
1256 corrections_snapshot = dict(self.state.corrections)
1258 if self.queue.completed_count > 0:
1259 total_issue_time = sum(t.get("total", 0) for t in timing_snapshot.values())
1260 if total_issue_time > 0:
1261 speedup = total_issue_time / total_time
1262 self.logger.info(f"Estimated speedup: {speedup:.2f}x")
1264 if self.queue.failed_ids:
1265 self.logger.info("")
1266 self.logger.warning("Failed issues:")
1267 for issue_id in self.queue.failed_ids:
1268 self.logger.warning(f" - {issue_id}")
1270 # Report interrupted issues separately (ENH-036)
1271 if self._interrupted_issues:
1272 self.logger.info("")
1273 self.logger.info(f"Interrupted: {len(self._interrupted_issues)} (can retry)")
1274 for issue_id in self._interrupted_issues:
1275 self.logger.info(f" - {issue_id}")
1277 # Report feature branches with actual per-branch state (ENH-665, BUG-2172)
1278 if self._pr_ready_branches:
1279 self.logger.info("")
1280 self.logger.info(f"Feature branches: {len(self._pr_ready_branches)} branch(es)")
1281 for issue_id, state in self._pr_ready_branches.items():
1282 branch = state["branch_name"]
1283 if state.get("pr_url"):
1284 self.logger.info(
1285 f" - {issue_id}: {branch} — pushed + PR opened: {state['pr_url']}"
1286 )
1287 elif state.get("pushed"):
1288 self.logger.info(f" - {issue_id}: {branch} — pushed (PR skipped)")
1289 else:
1290 self.logger.info(f" - {issue_id}: {branch} — local-only branch retained")
1292 # Report correction statistics for quality tracking (ENH-010)
1293 if corrections_snapshot:
1294 total_corrected = len(corrections_snapshot)
1295 total_issues = self.queue.completed_count + self.queue.failed_count
1296 correction_rate = (total_corrected / total_issues * 100) if total_issues > 0 else 0
1297 self.logger.info("")
1298 self.logger.info(
1299 f"Auto-corrections: {total_corrected}/{total_issues} ({correction_rate:.1f}%)"
1300 )
1302 # Group corrections by category (ENH-010 fourth fix)
1303 from collections import Counter, defaultdict
1305 all_corrections: list[str] = []
1306 by_category: dict[str, int] = defaultdict(int)
1307 for corrections in corrections_snapshot.values():
1308 all_corrections.extend(corrections)
1309 for correction in corrections:
1310 # Extract category from [category] prefix if present
1311 if correction.startswith("[") and "]" in correction:
1312 category = correction[1 : correction.index("]")]
1313 by_category[category] += 1
1314 else:
1315 by_category["uncategorized"] += 1
1317 # Log corrections by type/category
1318 if by_category:
1319 self.logger.info("Corrections by type:")
1320 for category, count in sorted(by_category.items(), key=lambda x: -x[1]):
1321 self.logger.info(f" - {category}: {count}")
1323 # Log most common individual corrections
1324 if all_corrections:
1325 common = Counter(all_corrections).most_common(3)
1326 self.logger.info("Most common corrections:")
1327 for correction, count in common:
1328 # Truncate long correction descriptions
1329 display = correction[:60] + "..." if len(correction) > 60 else correction
1330 self.logger.info(f" - {display}: {count}")
1332 # Report stash pop warnings (local changes need manual recovery)
1333 stash_warnings = self.merge_coordinator.stash_pop_failures
1334 if stash_warnings:
1335 self.logger.info("")
1336 self.logger.warning("Stash recovery warnings (local changes need manual restoration):")
1337 for issue_id, message in stash_warnings.items():
1338 self.logger.warning(f" - {issue_id}: {message}")
1339 self.logger.warning("")
1340 self.logger.warning(
1341 "To recover: Run 'git stash list' to find your changes, "
1342 "then 'git stash pop' or 'git stash apply stash@{N}'"
1343 )
1345 def _complete_issue_lifecycle_if_needed(
1346 self, issue_id: str, terminal_status: str = "done"
1347 ) -> bool:
1348 """Complete issue lifecycle by writing status to frontmatter.
1350 Args:
1351 issue_id: ID of the issue to complete
1352 terminal_status: Status to write — ``"done"`` for auto-merge,
1353 ``"in_progress"`` for feature-branch hold (ENH-2182)
1355 Returns:
1356 True if lifecycle was completed (or already complete), False on error
1357 """
1358 # TODO(ENH-1686): parallel-path close events not yet live-written
1359 info = self._issue_info_by_id.get(issue_id)
1360 if not info:
1361 self.logger.warning(f"No issue info found for {issue_id}")
1362 return False
1364 original_path = info.path
1366 if not original_path.exists():
1367 return True
1369 self.logger.info(f"Completing lifecycle for {issue_id} (frontmatter status update)")
1371 try:
1372 content = original_path.read_text()
1374 # Add resolution section if not already present
1375 if "## Resolution" not in content:
1376 action = self.br_config.get_category_action(info.issue_type)
1377 if terminal_status == "in_progress":
1378 status_label = "Branch ready, awaiting PR merge"
1379 impl_note = "Feature branch created; PR pending review and merge"
1380 else:
1381 status_label = "Completed (parallel merge fallback)"
1382 impl_note = "Merged from parallel worker branch"
1383 resolution = f"""
1385---
1387## Resolution
1389- **Action**: {action}
1390- **Completed**: {datetime.now(UTC).strftime("%Y-%m-%d")}
1391- **Status**: {status_label}
1392- **Implementation**: {impl_note}
1394### Changes Made
1395- See git history for implementation details
1397### Verification Results
1398- Work verification passed before merge
1400### Commits
1401- See `git log --oneline` for merge commit details
1402"""
1403 content += resolution
1405 fm_updates: dict[str, Any] = {"status": terminal_status}
1406 if terminal_status == "done":
1407 fm_updates["completed_at"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
1408 content = update_frontmatter(content, fm_updates)
1409 original_path.write_text(content)
1410 append_session_log_entry(original_path, "ll-parallel")
1412 # Stage and commit
1413 self._git_lock.run(
1414 ["add", "-A"],
1415 cwd=self.repo_path,
1416 )
1418 action = self.br_config.get_category_action(info.issue_type)
1419 if terminal_status == "in_progress":
1420 lifecycle_note = "Feature branch ready — status: in_progress (awaiting PR merge)"
1421 else:
1422 lifecycle_note = "Parallel merge fallback — status: done written to frontmatter"
1423 commit_msg = f"""{action}({info.issue_type}): complete {issue_id} lifecycle
1425{lifecycle_note}
1427Issue: {issue_id}
1428Type: {info.issue_type}
1429Title: {info.title}
1430"""
1431 commit_result = self._git_lock.run(
1432 ["commit", "-m", commit_msg],
1433 cwd=self.repo_path,
1434 )
1436 if commit_result.returncode != 0:
1437 if "nothing to commit" not in commit_result.stdout.lower():
1438 self.logger.warning(f"git commit failed: {commit_result.stderr}")
1439 else:
1440 commit_hash_match = re.search(r"\[[\w-]+\s+([a-f0-9]+)\]", commit_result.stdout)
1441 if commit_hash_match:
1442 self.logger.success(
1443 f"Completed lifecycle for {issue_id}: {commit_hash_match.group(1)}"
1444 )
1445 else:
1446 self.logger.success(f"Completed lifecycle for {issue_id}")
1448 return True
1450 except Exception as e:
1451 self.logger.error(f"Failed to complete lifecycle for {issue_id}: {e}")
1452 return False
1454 def _cleanup(self) -> None:
1455 """Clean up resources."""
1456 self.logger.info("Cleaning up...")
1458 # Save final state (force=True bypasses throttle to ensure shutdown state is persisted)
1459 self._save_state(force=True)
1461 # Shutdown components
1462 self.worker_pool.shutdown(wait=True)
1463 self.merge_coordinator.shutdown(wait=True, timeout=30)
1465 # Flush transports regardless of interrupt state so events are not lost.
1466 if self._event_bus is not None:
1467 self._event_bus.close_transports()
1469 # Clean up worktrees if not interrupted
1470 if not self._shutdown_requested:
1471 self.worker_pool.cleanup_all_worktrees()