Coverage for little_loops / parallel / merge_coordinator.py: 10%
465 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Merge coordinator for sequential integration of parallel worker changes.
3Handles merging completed worker branches back to main with conflict detection
4and automatic retry capability.
5"""
7from __future__ import annotations
9import shutil
10import subprocess
11import threading
12import time
13from pathlib import Path
14from queue import Empty, Queue
15from typing import TYPE_CHECKING
17from little_loops.parallel.git_lock import GitLock
18from little_loops.parallel.types import (
19 MergeRequest,
20 MergeStatus,
21 ParallelConfig,
22 WorkerResult,
23)
25if TYPE_CHECKING:
26 from little_loops.logger import Logger
29class MergeCoordinator:
30 """Sequential merge queue with conflict handling.
32 Processes merge requests one at a time to avoid conflicts. Supports
33 automatic rebase and retry on merge failures. Handles uncommitted local
34 changes by stashing them before merge operations.
36 Example:
37 >>> coordinator = MergeCoordinator(config, logger, repo_path)
38 >>> coordinator.start()
39 >>> coordinator.queue_merge(worker_result)
40 >>> # ... later ...
41 >>> coordinator.shutdown()
42 """
44 def __init__(
45 self,
46 config: ParallelConfig,
47 logger: Logger,
48 repo_path: Path | None = None,
49 git_lock: GitLock | None = None,
50 ) -> None:
51 """Initialize the merge coordinator.
53 Args:
54 config: Parallel processing configuration
55 logger: Logger for merge output
56 repo_path: Path to the git repository (default: current directory)
57 git_lock: Shared lock for git operations (created if not provided)
58 """
59 self.config = config
60 self.logger = logger
61 self.repo_path = repo_path or Path.cwd()
62 self._git_lock = git_lock or GitLock(logger)
63 self._queue: Queue[MergeRequest] = Queue()
64 self._thread: threading.Thread | None = None
65 self._shutdown_event = threading.Event()
66 self._merged: list[str] = []
67 self._failed: dict[str, str] = {}
68 self._lock = threading.Lock()
69 self._stash_active = False # Track if we have an active stash
70 self._consecutive_failures = 0 # Circuit breaker counter
71 self._paused = False # Set when circuit breaker trips
72 self._assume_unchanged_active = False # Track if state file is marked assume-unchanged
73 self._stash_pop_failures: dict[str, str] = {} # issue_id -> failure message
74 self._current_issue_id: str | None = (
75 None # Track current issue for stash failure attribution
76 )
77 self._problematic_commits: set[str] = (
78 set()
79 ) # Track commits causing repeated rebase conflicts
81 def start(self) -> None:
82 """Start the merge coordinator background thread."""
83 if self._thread is not None and self._thread.is_alive():
84 return
86 self._shutdown_event.clear()
87 self._thread = threading.Thread(
88 target=self._merge_loop,
89 name="merge-coordinator",
90 daemon=True,
91 )
92 self._thread.start()
93 self.logger.info("Merge coordinator started")
95 def shutdown(self, wait: bool = True, timeout: float = 30.0) -> None:
96 """Shutdown the merge coordinator.
98 Args:
99 wait: Whether to wait for pending merges to complete
100 timeout: Maximum time to wait for shutdown
101 """
102 if self._thread is None:
103 return
105 self._shutdown_event.set()
107 if wait and self._thread.is_alive():
108 self._thread.join(timeout=timeout)
110 self._thread = None
111 self.logger.info("Merge coordinator stopped")
113 def queue_merge(self, worker_result: WorkerResult) -> None:
114 """Queue a worker result for merging.
116 Args:
117 worker_result: Result from a completed worker
118 """
119 request = MergeRequest(worker_result=worker_result)
120 self._queue.put(request)
121 self.logger.info(
122 f"Queued merge for {worker_result.issue_id} (branch: {worker_result.branch_name})"
123 )
125 def _stash_local_changes(self) -> bool:
126 """Stash any uncommitted tracked changes in the main repo.
128 Only stashes tracked file modifications. Untracked files are not stashed
129 because git stash pathspec exclusions don't work reliably with -u flag.
130 Untracked file conflicts during merge are handled by _handle_untracked_conflict.
132 The following are explicitly excluded from stashing:
133 1. State file - managed by orchestrator and continuously updated
134 2. Lifecycle file moves - issue files being moved to completed/ directory
135 3. Files in completed directory - lifecycle-managed files
136 4. Claude Code context state file - managed by Claude Code externally
138 These exclusions prevent stash pop conflicts after merge, since the merge
139 may change HEAD and create conflicts with stashed rename operations.
141 Returns:
142 True if changes were stashed, False if working tree was clean
143 """
144 state_file_path = Path(self.config.state_file)
145 state_file_str = str(state_file_path)
146 state_file_name = state_file_path.name
148 # Check if there are any tracked changes to stash.
149 # We only look at tracked files (exclude untracked with grep -v '??')
150 # since we can only reliably stash tracked changes.
151 status_result = self._git_lock.run(
152 ["status", "--porcelain"],
153 cwd=self.repo_path,
154 timeout=30,
155 )
157 # Filter to only tracked changes (lines not starting with ??)
158 # and exclude orchestrator-managed files to prevent stash pop conflicts
159 tracked_changes = []
160 files_to_stash = []
161 # Note: Don't use .strip() on the full output - it removes leading spaces
162 # from the first line which are significant in git status porcelain format
163 for line in status_result.stdout.splitlines():
164 if not line or line.startswith("??"):
165 continue
166 # Skip lifecycle file moves (issue files moved to completed/)
167 # These are managed by orchestrator and cause stash pop conflicts
168 if self._is_lifecycle_file_move(line):
169 self.logger.debug(f"Skipping lifecycle file move from stash: {line}")
170 continue
171 # Extract file path from porcelain format (XY filename or XY -> filename for renames)
172 # Format: XY filename or XY old -> new (XY is exactly 2 chars + 1 space)
173 file_path = line[3:].split(" -> ")[-1].strip()
174 if file_path == state_file_str or file_path.endswith(state_file_name):
175 continue # Skip state file - orchestrator manages it independently
176 # Skip files in completed/deferred directory - these are lifecycle-managed.
177 # Handle both .issues/completed/ (with dot) and issues/completed/ (without dot).
178 # Use startswith to anchor the match and avoid false positives on paths like
179 # "my-issues/completed/file.py" that contain the substring but are unrelated.
180 if (
181 file_path.startswith(".issues/completed/")
182 or file_path.startswith(".issues/deferred/")
183 or file_path.startswith("issues/completed/")
184 or file_path.startswith("issues/deferred/")
185 ):
186 self.logger.debug(
187 f"Skipping completed/deferred directory file from stash: {file_path}"
188 )
189 continue
190 # Skip Claude Code context state file - managed externally
191 if file_path.endswith("ll-context-state.json"):
192 self.logger.debug(f"Skipping Claude context state file from stash: {file_path}")
193 continue
194 tracked_changes.append(line)
195 files_to_stash.append(file_path)
197 if not files_to_stash:
198 return False # No tracked changes to stash (excluding state file)
200 # Log files to be stashed for debugging
201 self.logger.debug(f"Tracked files to stash: {tracked_changes[:10]}")
203 # Stash only specific files, explicitly excluding the state file.
204 # Using explicit file list avoids race conditions where the orchestrator
205 # modifies the state file between a checkout and stash-all operation.
206 # Note: gitignored files are never stashed anyway.
207 stash_result = self._git_lock.run(
208 [
209 "stash",
210 "push",
211 "-m",
212 "ll-parallel: auto-stash before merge",
213 "--",
214 *files_to_stash,
215 ],
216 cwd=self.repo_path,
217 timeout=30,
218 )
220 if stash_result.returncode == 0:
221 self._stash_active = True
222 self.logger.info("Stashed local changes before merge")
223 return True
225 self.logger.error(f"Failed to stash local changes: {stash_result.stderr}")
226 return False
228 def _pop_stash(self) -> bool:
229 """Restore stashed changes if any were stashed.
231 Important: This method preserves the merge even if stash pop fails.
232 We never reset --hard HEAD here because that would undo a successful merge.
234 Returns:
235 True if stash was successfully popped or no stash was active,
236 False if pop failed (stash is left for manual recovery).
237 """
238 if not self._stash_active:
239 return True
241 pop_result = self._git_lock.run(
242 ["stash", "pop"],
243 cwd=self.repo_path,
244 timeout=30,
245 )
247 if pop_result.returncode != 0:
248 self.logger.warning(f"Failed to pop stash: {pop_result.stderr.strip()}")
250 # Check if it's a conflict issue - in that case, stash pop may have
251 # partially applied. We need to clean up the index but preserve the merge.
252 status_result = self._git_lock.run(
253 ["status", "--porcelain"],
254 cwd=self.repo_path,
255 timeout=30,
256 )
258 # Check for unmerged entries from the stash pop attempt
259 unmerged_prefixes = ("UU", "AA", "DD", "AU", "UA", "DU", "UD")
260 has_unmerged = any(
261 line[:2] in unmerged_prefixes
262 for line in status_result.stdout.splitlines()
263 if len(line) >= 2
264 )
266 if has_unmerged:
267 # Clean up the conflicted stash pop without affecting the merge
268 # Use checkout to restore conflicted files to their post-merge state
269 self._git_lock.run(
270 ["checkout", "--theirs", "."],
271 cwd=self.repo_path,
272 timeout=30,
273 )
274 self._git_lock.run(
275 ["reset", "HEAD"],
276 cwd=self.repo_path,
277 timeout=30,
278 )
279 self.logger.info("Cleaned up conflicted stash pop, merge preserved")
281 # Leave the stash intact for manual recovery
282 self._stash_active = False
284 # Record this failure for reporting in final summary
285 if self._current_issue_id:
286 with self._lock:
287 self._stash_pop_failures[self._current_issue_id] = (
288 "Local changes could not be restored after merge. "
289 "Run 'git stash list' and 'git stash pop' to recover manually."
290 )
292 self.logger.warning(
293 "Stash could not be restored - your changes are saved in 'git stash list'. "
294 "Run 'git stash show' to view and 'git stash pop' to retry manually."
295 )
296 return False
298 self._stash_active = False
299 self.logger.info("Restored stashed local changes")
300 return True
302 def _mark_state_file_assume_unchanged(self) -> bool:
303 """Mark the state file as assume-unchanged to prevent git from seeing modifications.
305 This allows git pull --rebase to proceed even when the state file is modified,
306 since the orchestrator continuously updates it during processing.
308 Returns:
309 True if successfully marked, False otherwise
310 """
311 state_file = str(self.config.state_file)
313 # Check if file exists and is tracked
314 ls_files = self._git_lock.run(
315 ["ls-files", state_file],
316 cwd=self.repo_path,
317 timeout=10,
318 )
320 if not ls_files.stdout.strip():
321 # File not tracked, nothing to do
322 return True
324 result = self._git_lock.run(
325 ["update-index", "--assume-unchanged", state_file],
326 cwd=self.repo_path,
327 timeout=10,
328 )
330 if result.returncode == 0:
331 self._assume_unchanged_active = True
332 self.logger.debug(f"Marked {state_file} as assume-unchanged")
333 return True
335 self.logger.warning(f"Failed to mark state file assume-unchanged: {result.stderr}")
336 return False
338 def _restore_state_file_tracking(self) -> bool:
339 """Restore normal tracking for the state file.
341 Returns:
342 True if successfully restored, False otherwise
343 """
344 if not self._assume_unchanged_active:
345 return True
347 state_file = str(self.config.state_file)
349 result = self._git_lock.run(
350 ["update-index", "--no-assume-unchanged", state_file],
351 cwd=self.repo_path,
352 timeout=10,
353 )
355 self._assume_unchanged_active = False
357 if result.returncode != 0:
358 self.logger.warning(f"Failed to restore state file tracking: {result.stderr}")
359 return False
361 self.logger.debug(f"Restored tracking for {state_file}")
362 return True
364 def _is_lifecycle_file_move(self, porcelain_line: str) -> bool:
365 """Check if a porcelain status line represents a lifecycle file move.
367 Lifecycle file moves are issue files being moved to completed/ directory.
368 These are managed by the orchestrator and should not be stashed, as they
369 will conflict with the merge when popping.
371 Args:
372 porcelain_line: A line from `git status --porcelain` output
374 Returns:
375 True if this is a lifecycle file move that should be excluded from stash
376 """
377 # Rename entries have format: R old_path -> new_path
378 if not porcelain_line.startswith("R"):
379 return False
381 # Check if it's a move to completed/ directory
382 if " -> " not in porcelain_line:
383 return False
385 # Extract destination path (after " -> ")
386 parts = porcelain_line[3:].split(" -> ")
387 if len(parts) != 2:
388 return False
390 dest_path = parts[1].strip()
392 # Check if destination is in completed or deferred directory (with or without dot prefix).
393 # Use startswith to anchor the match to the path root, preventing false positives on
394 # paths like "my-issues/completed/file.py" that contain the substring but are unrelated.
395 return (
396 dest_path.startswith(".issues/completed/")
397 or dest_path.startswith(".issues/deferred/")
398 or dest_path.startswith("issues/completed/")
399 or dest_path.startswith("issues/deferred/")
400 )
402 def _commit_pending_lifecycle_moves(self) -> bool:
403 """Commit any uncommitted lifecycle file moves.
405 Lifecycle file moves (issue files moved to completed/) are excluded from
406 stashing to prevent stash pop conflicts. However, if they remain uncommitted
407 when a merge starts, they will block the merge. This method commits any such
408 pending moves before the merge proceeds.
410 Returns:
411 True if any lifecycle moves were committed or none existed,
412 False if commit failed
413 """
414 # Check for lifecycle file moves in git status
415 status_result = self._git_lock.run(
416 ["status", "--porcelain"],
417 cwd=self.repo_path,
418 timeout=30,
419 )
421 lifecycle_moves = []
422 for line in status_result.stdout.splitlines():
423 if self._is_lifecycle_file_move(line):
424 lifecycle_moves.append(line)
426 if not lifecycle_moves:
427 return True
429 self.logger.info(
430 f"Found {len(lifecycle_moves)} uncommitted lifecycle file move(s), committing..."
431 )
433 # Stage all changes (lifecycle moves should already be staged from git mv,
434 # but add -A ensures any associated content changes are included)
435 self._git_lock.run(
436 ["add", "-A"],
437 cwd=self.repo_path,
438 timeout=30,
439 )
441 # Commit the lifecycle moves
442 commit_result = self._git_lock.run(
443 [
444 "commit",
445 "-m",
446 "chore(issues): commit pending lifecycle file moves\n\n"
447 "Auto-committed before merge to prevent conflicts.",
448 ],
449 cwd=self.repo_path,
450 timeout=30,
451 )
453 if commit_result.returncode != 0:
454 if "nothing to commit" in commit_result.stdout.lower():
455 # This shouldn't happen since we detected moves, but handle gracefully
456 self.logger.debug("No changes to commit despite detecting lifecycle moves")
457 return True
458 self.logger.error(f"Failed to commit lifecycle moves: {commit_result.stderr}")
459 return False
461 self.logger.info("Committed pending lifecycle file moves")
462 return True
464 def _is_local_changes_error(self, error_output: str) -> bool:
465 """Check if the error is due to uncommitted local changes.
467 Args:
468 error_output: The stderr/stdout from the failed git command
470 Returns:
471 True if the error indicates local changes would be overwritten
472 """
473 indicators = [
474 "Your local changes to the following files would be overwritten",
475 "Please commit your changes or stash them before you merge",
476 "error: cannot pull with rebase: You have unstaged changes",
477 ]
478 return any(indicator in error_output for indicator in indicators)
480 def _is_untracked_files_error(self, error_output: str) -> bool:
481 """Check if the error is due to untracked files blocking merge.
483 Args:
484 error_output: The stderr/stdout from the failed git command
486 Returns:
487 True if the error indicates untracked files would be overwritten
488 """
489 indicators = [
490 "untracked working tree files would be overwritten by merge",
491 "Please move or remove them before you merge",
492 ]
493 return any(indicator in error_output for indicator in indicators)
495 def _is_index_error(self, error_output: str) -> bool:
496 """Check if the error is due to a corrupted git index.
498 Args:
499 error_output: The stderr/stdout from the failed git command
501 Returns:
502 True if the error indicates index problems
503 """
504 indicators = [
505 "you need to resolve your current index first",
506 "fatal: cannot do a partial commit during a merge",
507 "error: you have not concluded your merge",
508 ]
509 return any(indicator in error_output for indicator in indicators)
511 def _is_rebase_in_progress(self) -> bool:
512 """Check if a rebase is currently in progress.
514 Returns:
515 True if rebase is in progress (rebase-merge or rebase-apply exists)
516 """
517 rebase_merge = self.repo_path / ".git" / "rebase-merge"
518 rebase_apply = self.repo_path / ".git" / "rebase-apply"
519 return rebase_merge.exists() or rebase_apply.exists()
521 def _abort_rebase_if_in_progress(self) -> bool:
522 """Abort any in-progress rebase operation.
524 Returns:
525 True if rebase was aborted or none was in progress,
526 False if abort failed
527 """
528 if not self._is_rebase_in_progress():
529 return True
531 self.logger.warning("Detected rebase in progress, aborting...")
532 abort_result = self._git_lock.run(
533 ["rebase", "--abort"],
534 cwd=self.repo_path,
535 timeout=30,
536 )
538 if abort_result.returncode != 0:
539 self.logger.error(f"Failed to abort rebase: {abort_result.stderr}")
540 # Force hard reset as last resort
541 return self._attempt_hard_reset()
543 self.logger.info("Aborted incomplete rebase from pull")
544 return True
546 def _is_unmerged_files_error(self, error_output: str) -> bool:
547 """Check if the error is due to pre-existing unmerged files.
549 Args:
550 error_output: The stderr/stdout from the failed git command
552 Returns:
553 True if the error indicates unmerged files blocking the operation
554 """
555 indicators = [
556 "you have unmerged files",
557 "Merging is not possible because you have unmerged files",
558 "fix conflicts and then commit the result",
559 ]
560 return any(indicator in error_output for indicator in indicators)
562 def _detect_conflict_commit(self, error_output: str) -> str | None:
563 """Extract commit hash from rebase conflict output.
565 Looks for patterns like:
566 - "dropping ae3b85ec1cac501058f6e5da362be37be1c99801 feat(ai): add stall detectio"
568 Args:
569 error_output: The stderr/stdout from the failed git pull --rebase
571 Returns:
572 The 40-character commit hash if found, None otherwise
573 """
574 import re
576 # Pattern: "dropping <40-char-hash>" followed by space and message
577 # Match only full 40-char hashes to avoid false positives
578 match = re.search(r"dropping\s+([a-f0-9]{40})\s+", error_output, re.IGNORECASE)
579 return match.group(1) if match else None
581 def _check_and_recover_index(self) -> bool:
582 """Check git index health and attempt recovery if needed.
584 Returns:
585 True if index is healthy or was recovered, False if unrecoverable
586 """
587 # Check if we're in the middle of a merge
588 merge_head = self.repo_path / ".git" / "MERGE_HEAD"
589 if merge_head.exists():
590 self.logger.warning("Detected incomplete merge, aborting...")
591 abort_result = self._git_lock.run(
592 ["merge", "--abort"],
593 cwd=self.repo_path,
594 timeout=30,
595 )
596 if abort_result.returncode != 0:
597 self.logger.error(f"Failed to abort merge: {abort_result.stderr}")
598 return False
599 self.logger.info("Aborted incomplete merge")
601 # Check if we're in the middle of a rebase
602 rebase_dir = self.repo_path / ".git" / "rebase-merge"
603 rebase_apply = self.repo_path / ".git" / "rebase-apply"
604 if rebase_dir.exists() or rebase_apply.exists():
605 self.logger.warning("Detected incomplete rebase, aborting...")
606 abort_result = self._git_lock.run(
607 ["rebase", "--abort"],
608 cwd=self.repo_path,
609 timeout=30,
610 )
611 if abort_result.returncode != 0:
612 self.logger.error(f"Failed to abort rebase: {abort_result.stderr}")
613 return False
614 self.logger.info("Aborted incomplete rebase")
615 # Force reset after rebase abort - the abort can leave index in dirty state
616 # This is defensive since unmerged file detection below may not trigger
617 if not self._attempt_hard_reset():
618 return False
620 # Check for unmerged files in the index (UU, AA, DD, AU, UA, DU, UD prefixes)
621 # These can persist even after merge --abort in some edge cases
622 status_result = self._git_lock.run(
623 ["status", "--porcelain"],
624 cwd=self.repo_path,
625 timeout=30,
626 )
628 if status_result.returncode != 0:
629 self.logger.error(f"git status failed: {status_result.stderr}")
630 return self._attempt_hard_reset()
632 # Debug logging to diagnose unmerged detection issues
633 if status_result.stdout.strip():
634 self.logger.debug(f"Git status output: {status_result.stdout[:500]}")
636 # Check for unmerged entries (first two chars indicate index/worktree status)
637 # Unmerged states: UU (both modified), AA (both added), DD (both deleted),
638 # AU/UA (added by us/them), DU/UD (deleted by us/them)
639 unmerged_prefixes = ("UU", "AA", "DD", "AU", "UA", "DU", "UD")
640 has_unmerged = any(
641 line[:2] in unmerged_prefixes
642 for line in status_result.stdout.splitlines()
643 if len(line) >= 2
644 )
646 if has_unmerged:
647 self.logger.warning("Detected unmerged files in index, resetting...")
648 if not self._attempt_hard_reset():
649 return False
650 self.logger.info("Cleared unmerged files from index")
652 # Final safety check - if MERGE_HEAD still exists, force reset
653 # This catches edge cases where abort succeeded but state is still dirty
654 if merge_head.exists():
655 self.logger.warning("MERGE_HEAD persists after recovery attempts, forcing reset")
656 if not self._attempt_hard_reset():
657 return False
659 return True
661 def _attempt_hard_reset(self) -> bool:
662 """Attempt a hard reset to recover from index issues.
664 Returns:
665 True if reset succeeded, False otherwise
666 """
667 self.logger.warning("Attempting hard reset to recover...")
668 reset_result = self._git_lock.run(
669 ["reset", "--hard", "HEAD"],
670 cwd=self.repo_path,
671 timeout=30,
672 )
673 if reset_result.returncode != 0:
674 self.logger.error("Hard reset failed - manual intervention required")
675 return False
676 self.logger.info("Hard reset successful")
677 return True
679 def _merge_loop(self) -> None:
680 """Background thread loop for processing merge requests."""
681 while not self._shutdown_event.is_set():
682 try:
683 # Wait for next merge request with timeout
684 try:
685 request = self._queue.get(timeout=1.0)
686 except Empty:
687 continue
689 # Process the merge
690 self._process_merge(request)
692 except Exception as e:
693 self.logger.error(f"Merge loop error: {e}")
694 time.sleep(1.0)
696 def _process_merge(self, request: MergeRequest) -> None:
697 """Process a single merge request.
699 Stashes any uncommitted local changes before attempting the merge,
700 and restores them afterward (regardless of success/failure).
702 Args:
703 request: Merge request to process
704 """
705 result = request.worker_result
706 with self._lock:
707 self._current_issue_id = result.issue_id
708 self.logger.info(f"Processing merge for {result.issue_id}")
709 had_local_changes = False
711 try:
712 # Circuit breaker check
713 if self._paused:
714 self.logger.warning(
715 f"Merge coordinator paused due to repeated failures. "
716 f"Skipping {result.issue_id}. Manual intervention required."
717 )
718 self._handle_failure(request, "Merge coordinator paused - circuit breaker tripped")
719 return
721 request.status = MergeStatus.IN_PROGRESS
723 # Health check: ensure git index is clean before proceeding
724 if not self._check_and_recover_index():
725 self._consecutive_failures += 1
726 if self._consecutive_failures >= 3:
727 self._paused = True
728 self.logger.error(
729 f"Circuit breaker tripped after {self._consecutive_failures} consecutive failures. "
730 "Merge coordinator paused. Manual recovery required."
731 )
732 self._handle_failure(request, "Git index recovery failed")
733 return
735 # Mark state file as assume-unchanged to prevent pull --rebase conflicts
736 # The orchestrator continuously updates the state file during processing
737 self._mark_state_file_assume_unchanged()
739 # Commit any uncommitted lifecycle file moves before stash/merge
740 # These are excluded from stash to prevent pop conflicts, so they must
741 # be committed to avoid blocking the merge (BUG-018 fix)
742 if not self._commit_pending_lifecycle_moves():
743 self._consecutive_failures += 1
744 if self._consecutive_failures >= 3:
745 self._paused = True
746 self.logger.error(
747 f"Circuit breaker tripped after {self._consecutive_failures} consecutive failures. "
748 "Merge coordinator paused. Manual recovery required."
749 )
750 self._handle_failure(request, "Failed to commit pending lifecycle moves")
751 return
753 # Stash any local changes before merge operations
754 had_local_changes = self._stash_local_changes()
756 # Ensure we're on base branch in the main repo
757 base = self.config.base_branch
758 remote = self.config.remote_name
759 checkout_result = self._git_lock.run(
760 ["checkout", base],
761 cwd=self.repo_path,
762 timeout=30,
763 )
765 if checkout_result.returncode != 0:
766 error_output = checkout_result.stderr + checkout_result.stdout
767 if self._is_local_changes_error(error_output):
768 # This shouldn't happen since we stashed, but handle it anyway
769 self.logger.warning(
770 "Checkout failed due to local changes despite stash attempt"
771 )
772 if self._is_index_error(error_output):
773 # Try recovery
774 if self._check_and_recover_index():
775 # Retry checkout
776 checkout_result = self._git_lock.run(
777 ["checkout", base],
778 cwd=self.repo_path,
779 timeout=30,
780 )
781 if checkout_result.returncode == 0:
782 self.logger.info("Recovered from index error, checkout succeeded")
783 else:
784 raise RuntimeError(
785 f"Failed to checkout {base} after recovery: "
786 f"{checkout_result.stderr}"
787 )
788 else:
789 raise RuntimeError(f"Failed to checkout {base}: {checkout_result.stderr}")
790 else:
791 raise RuntimeError(f"Failed to checkout {base}: {checkout_result.stderr}")
793 # Track if merge strategy was used during pull (for conflict handling)
794 used_merge_strategy = False
796 # Pull latest changes
797 pull_result = self._git_lock.run(
798 ["pull", "--rebase", remote, base],
799 cwd=self.repo_path,
800 timeout=60,
801 )
803 # Handle pull failures
804 if pull_result.returncode != 0:
805 error_output = pull_result.stderr + pull_result.stdout
807 # Check if rebase conflicted - must abort before continuing
808 if self._is_rebase_in_progress():
809 conflict_commit = self._detect_conflict_commit(error_output)
811 if conflict_commit and conflict_commit in self._problematic_commits:
812 # Known problematic commit - use merge strategy instead
813 self.logger.info(
814 f"Repeated rebase conflict with {conflict_commit[:8]}, "
815 f"using merge strategy (git pull --no-rebase)"
816 )
817 if not self._abort_rebase_if_in_progress():
818 raise RuntimeError("Failed to recover from rebase conflict during pull")
820 # Attempt merge strategy pull
821 merge_pull_result = self._git_lock.run(
822 ["pull", "--no-rebase", remote, base],
823 cwd=self.repo_path,
824 timeout=60,
825 )
827 if merge_pull_result.returncode != 0:
828 self.logger.warning(
829 f"Merge strategy pull also failed: {merge_pull_result.stderr[:200]}"
830 )
831 # Continue anyway - merge may still work or fail appropriately
832 else:
833 self.logger.info(
834 f"Merge strategy pull succeeded for {conflict_commit[:8]}"
835 )
836 used_merge_strategy = True
838 else:
839 # First time seeing this conflict or couldn't extract commit
840 if conflict_commit:
841 self._problematic_commits.add(conflict_commit)
842 self.logger.warning(
843 f"New rebase conflict with {conflict_commit[:8]}, "
844 f"tracking for future merges (will use merge strategy on repeat)"
845 )
846 else:
847 self.logger.warning(
848 "Rebase conflict detected (could not extract commit hash), "
849 "tracking for future merges"
850 )
852 self.logger.warning(
853 f"Pull --rebase failed with conflicts: {error_output[:200]}"
854 )
856 if not self._abort_rebase_if_in_progress():
857 raise RuntimeError("Failed to recover from rebase conflict during pull")
858 # After aborting rebase, we're back to pre-pull state
859 # Continue without the pull - merge may still work or conflict
860 self.logger.info("Continuing without pull after rebase abort")
862 elif self._is_local_changes_error(error_output):
863 self.logger.warning(
864 f"Pull failed due to local changes, attempting re-stash: {error_output[:200]}"
865 )
866 # Re-stash any local changes that appeared during pull
867 if self._stash_local_changes():
868 self.logger.info("Re-stashed local changes after pull conflict")
869 had_local_changes = True
870 # For other pull failures, continue - merge will handle or fail
872 # Safety check: ensure no unmerged files before merge attempt
873 # This catches edge cases where previous operations left dirty state
874 if not self._check_and_recover_index():
875 raise RuntimeError(
876 "Git index has unresolved conflicts before merge - recovery failed"
877 )
879 # Attempt merge with no-ff
880 merge_result = self._git_lock.run(
881 [
882 "merge",
883 result.branch_name,
884 "--no-ff",
885 "-m",
886 f"feat: parallel merge {result.issue_id}\n\n"
887 f"Automated merge from parallel issue processing.",
888 ],
889 cwd=self.repo_path,
890 timeout=60,
891 )
893 if merge_result.returncode != 0:
894 error_output = merge_result.stderr + merge_result.stdout
896 # Check for local changes error (shouldn't happen after stash)
897 if self._is_local_changes_error(error_output):
898 self.logger.warning(
899 f"Merge blocked by local changes despite stash: {error_output[:200]}"
900 )
901 raise RuntimeError(f"Merge failed due to local changes: {error_output[:200]}")
903 # Check for untracked files blocking merge
904 if self._is_untracked_files_error(error_output):
905 self._handle_untracked_conflict(request, error_output)
906 return
908 # Check for merge conflict (including unmerged files from current merge)
909 # Unmerged files at this point are genuine conflicts from the current merge
910 # attempt, not leftover state from previous operations (those are cleaned up
911 # by _check_and_recover_index() at the start of _process_merge)
912 if self._is_unmerged_files_error(error_output) or "CONFLICT" in error_output:
913 self._handle_conflict(request, used_merge_strategy)
914 return
915 else:
916 raise RuntimeError(f"Merge failed: {merge_result.stderr}")
918 # Merge successful
919 self._finalize_merge(request)
921 except Exception as e:
922 self._consecutive_failures += 1
923 if self._consecutive_failures >= 3:
924 self._paused = True
925 self.logger.error(
926 f"Circuit breaker tripped after {self._consecutive_failures} consecutive failures. "
927 "Merge coordinator paused. Manual recovery required."
928 )
929 self._handle_failure(request, str(e))
931 finally:
932 # Always restore stashed changes
933 if had_local_changes:
934 self._pop_stash()
935 # Always restore state file tracking
936 self._restore_state_file_tracking()
937 # Clear current issue tracking
938 with self._lock:
939 self._current_issue_id = None
941 def _handle_conflict(self, request: MergeRequest, used_merge_strategy: bool = False) -> None:
942 """Handle a merge conflict with retry logic.
944 Args:
945 request: The merge request that conflicted
946 used_merge_strategy: If True, merge strategy was used during pull and
947 rebase retry should be skipped (rebase would fail on same conflicts)
948 """
949 result = request.worker_result
950 request.retry_count += 1
952 # Abort the failed merge
953 self._git_lock.run(
954 ["merge", "--abort"],
955 cwd=self.repo_path,
956 timeout=10,
957 )
959 # Skip rebase retry if merge strategy was used during pull (BUG-079)
960 # Rebase would fail on the same commits that caused the initial conflict
961 if used_merge_strategy:
962 self.logger.warning(
963 f"Merge conflict for {result.issue_id}, "
964 f"skipping rebase retry (merge strategy was used during pull)"
965 )
966 self._handle_failure(
967 request,
968 "Merge conflict - rebase not attempted (would fail on same conflicts "
969 "that required merge strategy)",
970 )
971 return
973 if request.retry_count <= self.config.max_merge_retries:
974 # Attempt rebase in the worktree
975 self.logger.warning(
976 f"Merge conflict for {result.issue_id}, "
977 f"attempting rebase (retry {request.retry_count}/{self.config.max_merge_retries})"
978 )
980 request.status = MergeStatus.RETRYING
982 # Check for and stash any unstaged changes in the worktree before rebase
983 worktree_status = subprocess.run(
984 ["git", "status", "--porcelain"],
985 cwd=result.worktree_path,
986 capture_output=True,
987 text=True,
988 timeout=30,
989 )
990 worktree_has_changes = bool(worktree_status.stdout.strip())
992 if worktree_has_changes:
993 self.logger.debug(
994 f"Stashing worktree changes before rebase: {worktree_status.stdout[:200]}"
995 )
996 stash_result = subprocess.run(
997 ["git", "stash", "push", "-m", "ll-parallel: auto-stash before rebase"],
998 cwd=result.worktree_path,
999 capture_output=True,
1000 text=True,
1001 timeout=30,
1002 )
1003 if stash_result.returncode != 0:
1004 self.logger.warning(f"Failed to stash worktree changes: {stash_result.stderr}")
1006 # Fetch latest base branch before rebase (BUG-180)
1007 # Use remote/base if fetch succeeds, fall back to base if no remote
1008 base = self.config.base_branch
1009 remote = self.config.remote_name
1010 fetch_result = subprocess.run(
1011 ["git", "fetch", remote, base],
1012 cwd=result.worktree_path,
1013 capture_output=True,
1014 text=True,
1015 timeout=60,
1016 )
1017 rebase_target = f"{remote}/{base}" if fetch_result.returncode == 0 else base
1019 # Rebase the branch onto latest base branch (BUG-180)
1020 rebase_result = subprocess.run(
1021 ["git", "rebase", rebase_target],
1022 cwd=result.worktree_path,
1023 capture_output=True,
1024 text=True,
1025 timeout=120,
1026 )
1028 if rebase_result.returncode == 0:
1029 # Rebase succeeded, restore stash if we made one, then retry merge
1030 if worktree_has_changes:
1031 pop_result = subprocess.run(
1032 ["git", "stash", "pop"],
1033 cwd=result.worktree_path,
1034 capture_output=True,
1035 text=True,
1036 timeout=30,
1037 )
1038 if pop_result.returncode != 0:
1039 self.logger.error(
1040 f"Stash pop failed for {result.issue_id} after rebase: "
1041 f"{pop_result.stderr.strip()}"
1042 )
1043 show_result = subprocess.run(
1044 ["git", "stash", "show", "-p"],
1045 cwd=result.worktree_path,
1046 capture_output=True,
1047 text=True,
1048 timeout=30,
1049 )
1050 if show_result.returncode == 0 and show_result.stdout:
1051 self.logger.warning(
1052 f"Dropping unrecoverable worktree stash for {result.issue_id}:\n"
1053 f"{show_result.stdout[:2000]}"
1054 )
1055 subprocess.run(
1056 ["git", "stash", "drop"],
1057 cwd=result.worktree_path,
1058 capture_output=True,
1059 timeout=10,
1060 )
1061 self._handle_failure(request, "Stash pop conflict after rebase")
1062 return
1063 self._queue.put(request)
1064 else:
1065 # Rebase also failed - abort and restore stash
1066 subprocess.run(
1067 ["git", "rebase", "--abort"],
1068 cwd=result.worktree_path,
1069 capture_output=True,
1070 timeout=10,
1071 )
1072 if worktree_has_changes:
1073 subprocess.run(
1074 ["git", "stash", "pop"],
1075 cwd=result.worktree_path,
1076 capture_output=True,
1077 timeout=30,
1078 )
1079 self._handle_failure(
1080 request,
1081 f"Rebase failed after merge conflict: {rebase_result.stderr}",
1082 )
1083 else:
1084 self._handle_failure(
1085 request,
1086 f"Merge conflict after {request.retry_count} retries",
1087 )
1089 def _handle_untracked_conflict(self, request: MergeRequest, error_output: str) -> None:
1090 """Handle untracked files that would be overwritten by merge.
1092 Backs up conflicting untracked files and retries the merge.
1094 Args:
1095 request: The merge request that failed
1096 error_output: Git error message containing file list
1097 """
1098 result = request.worker_result
1099 request.retry_count += 1
1101 if request.retry_count > self.config.max_merge_retries:
1102 self._handle_failure(
1103 request,
1104 f"Untracked file conflict after {request.retry_count} retries",
1105 )
1106 return
1108 # Parse conflicting files from error message
1109 # Format: "error: The following untracked working tree files would be overwritten..."
1110 # followed by file paths, then "Please move or remove them..."
1111 conflicting_files = []
1112 in_file_list = False
1113 for line in error_output.splitlines():
1114 line = line.strip()
1115 if "untracked working tree files would be overwritten" in line:
1116 in_file_list = True
1117 continue
1118 if "Please move or remove them" in line:
1119 in_file_list = False
1120 continue
1121 if in_file_list and line and not line.startswith("error:"):
1122 conflicting_files.append(line)
1124 if not conflicting_files:
1125 self._handle_failure(
1126 request,
1127 f"Could not parse conflicting files from: {error_output[:200]}",
1128 )
1129 return
1131 # Create backup directory
1132 backup_dir = self.repo_path / ".ll-backup" / result.issue_id
1133 backup_dir.mkdir(parents=True, exist_ok=True)
1135 # Move conflicting files to backup
1136 moved_files = []
1137 for file_path in conflicting_files:
1138 src = self.repo_path / file_path
1139 if src.exists():
1140 dst = backup_dir / file_path
1141 dst.parent.mkdir(parents=True, exist_ok=True)
1142 shutil.move(str(src), str(dst))
1143 moved_files.append(file_path)
1145 if moved_files:
1146 self.logger.info(
1147 f"Backed up {len(moved_files)} conflicting untracked file(s) to {backup_dir}"
1148 )
1150 # Retry the merge
1151 self.logger.warning(
1152 f"Untracked files conflict for {result.issue_id}, "
1153 f"retrying after backup (attempt {request.retry_count}/{self.config.max_merge_retries})"
1154 )
1155 request.status = MergeStatus.RETRYING
1156 self._queue.put(request)
1158 def _finalize_merge(self, request: MergeRequest) -> None:
1159 """Finalize a successful merge.
1161 Args:
1162 request: The completed merge request
1163 """
1164 result = request.worker_result
1165 request.status = MergeStatus.SUCCESS
1167 # Reset circuit breaker on success
1168 self._consecutive_failures = 0
1170 with self._lock:
1171 self._merged.append(result.issue_id)
1173 # Cleanup worktree and branch
1174 self._cleanup_worktree(result.worktree_path, result.branch_name)
1176 self.logger.success(f"Merged {result.issue_id} successfully")
1178 def _handle_failure(self, request: MergeRequest, error: str) -> None:
1179 """Handle a merge failure.
1181 Args:
1182 request: The failed merge request
1183 error: Error message describing the failure
1184 """
1185 result = request.worker_result
1186 request.status = MergeStatus.FAILED
1187 request.error = error
1189 with self._lock:
1190 self._failed[result.issue_id] = error
1192 self.logger.error(f"Merge failed for {result.issue_id}: {error}")
1194 def _cleanup_worktree(self, worktree_path: Path, branch_name: str) -> None:
1195 """Clean up a merged worktree and its branch.
1197 Args:
1198 worktree_path: Path to the worktree
1199 branch_name: Name of the branch to delete
1200 """
1201 if not worktree_path.exists():
1202 return
1204 # Remove worktree
1205 self._git_lock.run(
1206 ["worktree", "remove", "--force", str(worktree_path)],
1207 cwd=self.repo_path,
1208 timeout=30,
1209 )
1211 # Force delete directory if still exists
1212 if worktree_path.exists():
1213 shutil.rmtree(worktree_path, ignore_errors=True)
1215 # Delete the branch
1216 if branch_name.startswith("parallel/"):
1217 self._git_lock.run(
1218 ["branch", "-D", branch_name],
1219 cwd=self.repo_path,
1220 timeout=10,
1221 )
1223 @property
1224 def merged_ids(self) -> list[str]:
1225 """List of successfully merged issue IDs."""
1226 with self._lock:
1227 return list(self._merged)
1229 @property
1230 def failed_merges(self) -> dict[str, str]:
1231 """Mapping of failed issue IDs to error messages."""
1232 with self._lock:
1233 return dict(self._failed)
1235 @property
1236 def pending_count(self) -> int:
1237 """Number of pending merge requests."""
1238 return self._queue.qsize()
1240 @property
1241 def stash_pop_failures(self) -> dict[str, str]:
1242 """Mapping of issue IDs to stash pop failure messages.
1244 These represent issues where the merge succeeded but the user's
1245 local changes could not be automatically restored and need manual
1246 recovery via 'git stash pop'.
1247 """
1248 with self._lock:
1249 return dict(self._stash_pop_failures)
1251 def wait_for_completion(self, timeout: float | None = None) -> bool:
1252 """Wait for all pending merges to complete.
1254 Waits until both:
1255 1. The merge queue is empty (no pending requests)
1256 2. No merge is actively being processed (_current_issue_id is None)
1258 Args:
1259 timeout: Maximum time to wait (None = forever)
1261 Returns:
1262 True if all merges completed, False if timeout
1263 """
1264 start_time = time.time()
1265 while True:
1266 with self._lock:
1267 active = self._current_issue_id
1268 if self._queue.empty() and not active:
1269 return True
1270 if timeout and (time.time() - start_time) > timeout:
1271 return False
1272 time.sleep(0.5)