Coverage for little_loops / parallel / worker_pool.py: 11%
448 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""Worker pool for parallel issue processing with git worktree isolation.
3Each worker operates in an isolated git worktree, allowing concurrent issue
4processing without file conflicts.
5"""
7from __future__ import annotations
9import json
10import os
11import re
12import subprocess
13import sys
14import threading
15import time
16from collections.abc import Callable
17from concurrent.futures import Future, ThreadPoolExecutor
18from datetime import datetime
19from pathlib import Path
20from typing import TYPE_CHECKING, Any, cast
22from little_loops.output_parsing import parse_ready_issue_output
23from little_loops.parallel.git_lock import GitLock
24from little_loops.parallel.types import ParallelConfig, WorkerResult, WorkerStage
25from little_loops.subprocess_utils import (
26 detect_context_handoff,
27 read_continuation_prompt,
28)
29from little_loops.subprocess_utils import (
30 run_claude_command as _run_claude_base,
31)
32from little_loops.work_verification import EXCLUDED_DIRECTORIES, verify_work_was_done
34if TYPE_CHECKING:
35 from little_loops.config import BRConfig
36 from little_loops.issue_parser import IssueInfo
37 from little_loops.logger import Logger
40class WorkerPool:
41 """Thread pool for processing issues in isolated git worktrees.
43 Each worker:
44 1. Creates a dedicated git worktree and branch
45 2. Runs issue validation and implementation via Claude CLI
46 3. Commits changes locally
47 4. Returns results for merge coordination
49 Example:
50 >>> pool = WorkerPool(parallel_config, br_config, logger)
51 >>> pool.start()
52 >>> future = pool.submit(issue_info)
53 >>> result = future.result() # WorkerResult
54 >>> pool.shutdown()
55 """
57 def __init__(
58 self,
59 parallel_config: ParallelConfig,
60 br_config: BRConfig,
61 logger: Logger,
62 repo_path: Path | None = None,
63 git_lock: GitLock | None = None,
64 ) -> None:
65 """Initialize the worker pool.
67 Args:
68 parallel_config: Parallel processing configuration
69 br_config: Project configuration (for category actions)
70 logger: Logger for worker output
71 repo_path: Path to the git repository (default: current directory)
72 git_lock: Shared lock for git operations (created if not provided)
73 """
74 self.parallel_config = parallel_config
75 self.br_config = br_config
76 self.logger = logger
77 self.repo_path = repo_path or Path.cwd()
78 self._git_lock = git_lock or GitLock(logger)
79 self._executor: ThreadPoolExecutor | None = None
80 self._active_workers: dict[str, Future[WorkerResult]] = {}
81 # Track active subprocesses for forceful termination on shutdown
82 self._active_processes: dict[str, subprocess.Popen[str]] = {}
83 # Track active worktree paths to prevent cleanup while in use (BUG-142)
84 self._active_worktrees: set[Path] = set()
85 self._process_lock = threading.Lock()
86 # Track callbacks currently executing
87 self._pending_callbacks: set[str] = set()
88 self._callback_lock = threading.Lock()
89 # Shutdown tracking for interrupted worker detection (ENH-036)
90 self._shutdown_requested = False
91 self._terminated_during_shutdown: set[str] = set()
92 # Track worker processing stages for progress visibility (ENH-262)
93 self._worker_stages: dict[str, WorkerStage] = {}
95 def start(self) -> None:
96 """Start the worker pool."""
97 if self._executor is not None:
98 return
100 # Ensure worktree base directory exists
101 worktree_base = self.repo_path / self.parallel_config.worktree_base
102 worktree_base.mkdir(parents=True, exist_ok=True)
104 self._executor = ThreadPoolExecutor(
105 max_workers=self.parallel_config.max_workers,
106 thread_name_prefix="issue-worker",
107 )
108 self.logger.info(f"Worker pool started with {self.parallel_config.max_workers} workers")
110 def shutdown(self, wait: bool = True) -> None:
111 """Shutdown the worker pool.
113 Args:
114 wait: Whether to wait for pending tasks to complete
115 """
116 if self._executor is None:
117 return
119 self.logger.info("Shutting down worker pool...")
121 # First, terminate all active subprocesses to unblock worker threads
122 if not wait:
123 self.terminate_all_processes()
125 self._executor.shutdown(wait=wait)
126 self._executor = None
128 def set_shutdown_requested(self, value: bool = True) -> None:
129 """Set the shutdown flag.
131 Called by orchestrator during shutdown to enable tracking of
132 workers that are terminated due to shutdown vs. actual failures.
133 """
134 self._shutdown_requested = value
136 def terminate_all_processes(self) -> None:
137 """Forcefully terminate all active subprocesses.
139 Called when we need to abort workers immediately,
140 such as on timeout or shutdown.
141 """
142 with self._process_lock:
143 for issue_id, process in list(self._active_processes.items()):
144 if process.poll() is None: # Still running
145 self.logger.warning(
146 f"Terminating subprocess for {issue_id} (PID {process.pid})"
147 )
148 # Track issues terminated during shutdown for interrupted detection (ENH-036)
149 if self._shutdown_requested:
150 self._terminated_during_shutdown.add(issue_id)
151 try:
152 # Send SIGTERM first for graceful termination
153 process.terminate()
154 try:
155 process.wait(timeout=5)
156 except subprocess.TimeoutExpired:
157 # Force kill if SIGTERM didn't work
158 self.logger.warning(f"Force killing {issue_id} (PID {process.pid})")
159 process.kill()
160 process.wait(timeout=2)
161 except Exception as e:
162 self.logger.error(f"Failed to terminate {issue_id}: {e}")
163 self._active_processes.clear()
165 def submit(
166 self,
167 issue: IssueInfo,
168 on_complete: Callable[[WorkerResult], None] | None = None,
169 ) -> Future[WorkerResult]:
170 """Submit an issue for processing.
172 Args:
173 issue: Issue to process
174 on_complete: Optional callback when processing completes
176 Returns:
177 Future that will contain the WorkerResult
178 """
179 if self._executor is None:
180 raise RuntimeError("Worker pool not started")
182 future = self._executor.submit(self._process_issue, issue)
183 with self._process_lock:
184 self._active_workers[issue.issue_id] = future
186 if on_complete:
187 future.add_done_callback(
188 lambda f: self._handle_completion(f, on_complete, issue.issue_id)
189 )
191 return future
193 def _handle_completion(
194 self,
195 future: Future[WorkerResult],
196 callback: Callable[[WorkerResult], None],
197 issue_id: str,
198 ) -> None:
199 """Handle worker completion and invoke callback."""
200 with self._callback_lock:
201 self._pending_callbacks.add(issue_id)
202 try:
203 try:
204 result = future.result()
205 except Exception as e:
206 self.logger.error(f"Worker future failed for {issue_id}: {e}")
207 result = WorkerResult(
208 issue_id=issue_id,
209 success=False,
210 branch_name="",
211 worktree_path=Path(),
212 error=f"Worker future failed: {e}",
213 )
214 # Set final stage based on result (ENH-262)
215 if result.success:
216 self.set_worker_stage(issue_id, WorkerStage.COMPLETED)
217 elif result.interrupted:
218 self.set_worker_stage(issue_id, WorkerStage.INTERRUPTED)
219 else:
220 self.set_worker_stage(issue_id, WorkerStage.FAILED)
221 try:
222 callback(result)
223 except Exception as e:
224 self.logger.error(f"Worker completion callback failed for {issue_id}: {e}")
225 finally:
226 with self._callback_lock:
227 self._pending_callbacks.discard(issue_id)
229 def _process_issue(self, issue: IssueInfo) -> WorkerResult:
230 """Process a single issue in an isolated worktree.
232 Args:
233 issue: Issue to process
235 Returns:
236 WorkerResult with processing outcome
237 """
238 start_time = time.time()
239 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
240 if self.parallel_config.use_feature_branches:
241 from little_loops.issue_parser import slugify
243 branch_name = f"feature/{issue.issue_id.lower()}-{slugify(issue.title)}"
244 else:
245 branch_name = f"parallel/{issue.issue_id.lower()}-{timestamp}"
246 worktree_path = (
247 self.repo_path
248 / self.parallel_config.worktree_base
249 / f"worker-{issue.issue_id.lower()}-{timestamp}"
250 )
252 # Set initial stage for progress tracking (ENH-262)
253 self.set_worker_stage(issue.issue_id, WorkerStage.SETUP)
255 # Capture baseline of main repo status before worker starts
256 # Used to detect files incorrectly written to main repo
257 baseline_status = self._get_main_repo_baseline()
258 # Capture main HEAD SHA before worker starts to detect committed leaks
259 baseline_head_sha = self._get_main_head_sha()
261 try:
262 # Step 1: Create worktree with new branch
263 self._setup_worktree(worktree_path, branch_name)
265 # Register worktree as active to prevent cleanup while in use (BUG-142)
266 with self._process_lock:
267 self._active_worktrees.add(worktree_path)
269 # Update stage for progress tracking (ENH-262)
270 self.set_worker_stage(issue.issue_id, WorkerStage.VALIDATING)
272 # Step 2: Run ready-issue validation
273 ready_cmd = self.parallel_config.get_ready_command(issue.issue_id)
274 ready_result = self._run_claude_command(
275 ready_cmd,
276 worktree_path,
277 issue_id=issue.issue_id,
278 )
280 # Check if worker was terminated during shutdown (ENH-036)
281 if issue.issue_id in self._terminated_during_shutdown:
282 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
283 return WorkerResult(
284 issue_id=issue.issue_id,
285 success=False,
286 interrupted=True,
287 branch_name=branch_name,
288 worktree_path=worktree_path,
289 duration=time.time() - start_time,
290 error="Interrupted during shutdown",
291 stdout=ready_result.stdout,
292 stderr=ready_result.stderr,
293 )
295 if ready_result.returncode != 0:
296 return WorkerResult(
297 issue_id=issue.issue_id,
298 success=False,
299 branch_name=branch_name,
300 worktree_path=worktree_path,
301 duration=time.time() - start_time,
302 error=f"ready-issue failed: {ready_result.stderr}",
303 stdout=ready_result.stdout,
304 stderr=ready_result.stderr,
305 )
307 # Step 3: Parse ready-issue output and check verdict
308 ready_parsed = parse_ready_issue_output(ready_result.stdout)
310 # Handle CLOSE verdict - issue should not be implemented
311 if ready_parsed.get("should_close"):
312 return WorkerResult(
313 issue_id=issue.issue_id,
314 success=True, # Closure is a valid outcome
315 branch_name=branch_name,
316 worktree_path=worktree_path,
317 duration=time.time() - start_time,
318 should_close=True,
319 close_reason=ready_parsed.get("close_reason"),
320 close_status=ready_parsed.get("close_status"),
321 stdout=ready_result.stdout,
322 stderr=ready_result.stderr,
323 )
325 # Handle BLOCKED verdict - issue has open dependencies
326 if ready_parsed.get("is_blocked"):
327 return WorkerResult(
328 issue_id=issue.issue_id,
329 success=False,
330 was_blocked=True,
331 branch_name=branch_name,
332 worktree_path=worktree_path,
333 duration=time.time() - start_time,
334 error="ready-issue verdict: BLOCKED - open dependency detected",
335 stdout=ready_result.stdout,
336 stderr=ready_result.stderr,
337 )
339 # Handle NOT_READY verdict
340 if not ready_parsed["is_ready"]:
341 concerns = ready_parsed.get("concerns", [])
342 if concerns:
343 concern_msg = "; ".join(concerns)
344 elif ready_parsed["verdict"] == "UNKNOWN":
345 # For UNKNOWN verdicts, show a snippet of output for debugging
346 raw_out = (ready_result.stdout or "")[:200].strip()
347 concern_msg = (
348 f"Could not parse verdict. Output: {raw_out}..."
349 if raw_out
350 else "No output from ready-issue"
351 )
352 else:
353 concern_msg = "Issue not ready"
354 return WorkerResult(
355 issue_id=issue.issue_id,
356 success=False,
357 branch_name=branch_name,
358 worktree_path=worktree_path,
359 duration=time.time() - start_time,
360 error=f"ready-issue verdict: {ready_parsed['verdict']} - {concern_msg}",
361 stdout=ready_result.stdout,
362 stderr=ready_result.stderr,
363 )
365 # Track if issue was corrected (corrections stay in worktree)
366 was_corrected = ready_parsed.get("was_corrected", False)
367 corrections = ready_parsed.get("corrections", [])
369 # Update stage for progress tracking (ENH-262)
370 self.set_worker_stage(issue.issue_id, WorkerStage.IMPLEMENTING)
372 # Step 4: Get action from BRConfig
373 action = self.br_config.get_category_action(issue.issue_type)
375 # Step 5: Run manage-issue implementation (with continuation support)
376 manage_cmd = self.parallel_config.get_manage_command(
377 issue.issue_type, action, issue.issue_id
378 )
379 manage_result = self._run_with_continuation(
380 manage_cmd,
381 worktree_path,
382 issue_id=issue.issue_id,
383 )
385 # Update stage for progress tracking (ENH-262)
386 self.set_worker_stage(issue.issue_id, WorkerStage.VERIFYING)
388 # Check if worker was terminated during shutdown (ENH-036)
389 if issue.issue_id in self._terminated_during_shutdown:
390 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
391 return WorkerResult(
392 issue_id=issue.issue_id,
393 success=False,
394 interrupted=True,
395 branch_name=branch_name,
396 worktree_path=worktree_path,
397 duration=time.time() - start_time,
398 error="Interrupted during shutdown",
399 stdout=manage_result.stdout,
400 stderr=manage_result.stderr,
401 )
403 # Step 6: Get list of changed files in worktree
404 changed_files = self._get_changed_files(worktree_path)
406 # Step 8: Detect files leaked to main repo instead of worktree (unstaged)
407 leaked_files = self._detect_main_repo_leaks(issue.issue_id, baseline_status)
408 if leaked_files:
409 self.logger.warning(
410 f"{issue.issue_id} leaked {len(leaked_files)} file(s) to main repo: "
411 f"{leaked_files}"
412 )
413 # Clean up leaked files to prevent stash conflicts during merge.
414 # The actual work is preserved in the worktree branch.
415 self._cleanup_leaked_files(leaked_files)
417 # Step 8b: Detect commits made directly to main instead of worktree branch.
418 # If Claude committed to main (not the worktree), worktree will have no diff,
419 # causing work verification to fail. Attempt to recover by cherry-picking
420 # the leaked commits to the worktree and resetting main. (BUG-580)
421 committed_leaks = self._detect_committed_leaks(baseline_head_sha)
422 if committed_leaks:
423 self.logger.warning(
424 f"{issue.issue_id} committed {len(committed_leaks)} commit(s) directly "
425 f"to main instead of worktree: {[sha[:8] for sha in committed_leaks]}"
426 )
427 if not changed_files:
428 recovered = self._recover_committed_leaks(
429 committed_leaks, worktree_path, baseline_head_sha, issue.issue_id
430 )
431 if recovered:
432 changed_files = self._get_changed_files(worktree_path)
434 # Step 7: Verify actual work was done (after potential committed-leak recovery)
435 # Pass full filename for better doc-only keyword matching
436 issue_filename = issue.path.stem if issue.path else ""
437 work_verified, verification_error = self._verify_work_was_done(
438 changed_files, issue.issue_id, issue_filename
439 )
441 if manage_result.returncode != 0:
442 return WorkerResult(
443 issue_id=issue.issue_id,
444 success=False,
445 branch_name=branch_name,
446 worktree_path=worktree_path,
447 changed_files=changed_files,
448 leaked_files=leaked_files,
449 duration=time.time() - start_time,
450 error=f"manage-issue failed: {manage_result.stderr}",
451 stdout=manage_result.stdout,
452 stderr=manage_result.stderr,
453 )
455 if not work_verified:
456 return WorkerResult(
457 issue_id=issue.issue_id,
458 success=False,
459 branch_name=branch_name,
460 worktree_path=worktree_path,
461 changed_files=changed_files,
462 leaked_files=leaked_files,
463 duration=time.time() - start_time,
464 error=verification_error,
465 stdout=manage_result.stdout,
466 stderr=manage_result.stderr,
467 )
469 # Step 9: Update branch base before merge (BUG-180)
470 # Fetch origin/main and rebase to ensure branch is based on latest main
471 base_updated, base_error = self._update_branch_base(worktree_path, issue.issue_id)
473 # Update stage for progress tracking (ENH-262)
474 self.set_worker_stage(issue.issue_id, WorkerStage.MERGING)
476 if not base_updated:
477 return WorkerResult(
478 issue_id=issue.issue_id,
479 success=False,
480 branch_name=branch_name,
481 worktree_path=worktree_path,
482 changed_files=changed_files,
483 leaked_files=leaked_files,
484 duration=time.time() - start_time,
485 error=base_error,
486 stdout=manage_result.stdout,
487 stderr=manage_result.stderr,
488 )
490 return WorkerResult(
491 issue_id=issue.issue_id,
492 success=True,
493 branch_name=branch_name,
494 worktree_path=worktree_path,
495 changed_files=changed_files,
496 leaked_files=leaked_files,
497 duration=time.time() - start_time,
498 error=None,
499 stdout=manage_result.stdout,
500 stderr=manage_result.stderr,
501 was_corrected=was_corrected,
502 corrections=corrections,
503 )
505 except Exception as e:
506 return WorkerResult(
507 issue_id=issue.issue_id,
508 success=False,
509 branch_name=branch_name,
510 worktree_path=worktree_path,
511 duration=time.time() - start_time,
512 error=str(e),
513 )
514 finally:
515 # Unregister worktree as no longer active (BUG-142)
516 with self._process_lock:
517 self._active_worktrees.discard(worktree_path)
519 def _setup_worktree(self, worktree_path: Path, branch_name: str) -> None:
520 """Create a git worktree with a new branch.
522 Args:
523 worktree_path: Path for the new worktree
524 branch_name: Name of the new branch
525 """
526 from little_loops.worktree_utils import setup_worktree
528 setup_worktree(
529 repo_path=self.repo_path,
530 worktree_path=worktree_path,
531 branch_name=branch_name,
532 copy_files=self.parallel_config.worktree_copy_files,
533 logger=self.logger,
534 git_lock=self._git_lock,
535 )
537 # Verify model if --show-model flag is set (requires API call)
538 if self.parallel_config.show_model:
539 model = self._detect_worktree_model_via_api(worktree_path)
540 if model:
541 self.logger.info(f" Using model: {model}")
542 else:
543 self.logger.warning(" Could not detect Claude CLI model")
545 def _detect_worktree_model_via_api(self, worktree_path: Path) -> str | None:
546 """Detect the model Claude will use by making an API call.
548 Runs a minimal Claude command with JSON output and parses the modelUsage
549 field to verify settings.local.json is being respected.
551 Args:
552 worktree_path: Path to the worktree to test
554 Returns:
555 Model name (e.g., "claude-sonnet-4-20250514") or None if unable to detect
556 """
557 try:
558 # Set environment to keep Claude in the project working directory (BUG-007)
559 # This ensures the first Claude CLI invocation in the worktree has the same
560 # project root behavior as subsequent invocations via run_claude_command()
561 env = os.environ.copy()
562 env["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1"
564 # Use a minimal prompt that requires an API call to get modelUsage
565 result = subprocess.run(
566 [
567 "claude",
568 "-p",
569 "reply with just 'ok'",
570 "--output-format",
571 "json",
572 ],
573 cwd=worktree_path,
574 capture_output=True,
575 text=True,
576 timeout=30,
577 env=env,
578 )
579 if result.returncode == 0 and result.stdout.strip():
580 data: dict[str, Any] = json.loads(result.stdout.strip())
581 model_usage: dict[str, Any] = data.get("modelUsage", {})
582 # Return the first (primary) model from modelUsage
583 if model_usage:
584 return cast(str, next(iter(model_usage.keys())))
585 except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError):
586 pass
587 return None
589 def _cleanup_worktree(self, worktree_path: Path) -> None:
590 """Remove a git worktree and its associated branch.
592 Args:
593 worktree_path: Path to the worktree to remove
594 """
595 if not worktree_path.exists():
596 return
598 # Skip cleanup if worktree is actively in use by a running worker (BUG-142)
599 with self._process_lock:
600 if worktree_path in self._active_worktrees:
601 self.logger.warning(
602 f"Skipping cleanup of {worktree_path.name}: worktree is in active use"
603 )
604 return
606 # Only delete branches with the parallel/ prefix (legacy behavior for ll-parallel)
607 branch_result = subprocess.run(
608 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
609 cwd=worktree_path,
610 capture_output=True,
611 text=True,
612 )
613 branch_name = branch_result.stdout.strip() if branch_result.returncode == 0 else None
614 delete_branch = branch_name is not None and branch_name.startswith("parallel/")
616 from little_loops.worktree_utils import cleanup_worktree
618 cleanup_worktree(
619 worktree_path=worktree_path,
620 repo_path=self.repo_path,
621 logger=self.logger,
622 git_lock=self._git_lock,
623 delete_branch=delete_branch,
624 )
626 def _run_claude_command(
627 self,
628 command: str,
629 working_dir: Path,
630 issue_id: str | None = None,
631 ) -> subprocess.CompletedProcess[str]:
632 """Run a Claude CLI command with real-time output streaming.
634 Args:
635 command: The command to run (e.g., "/ll:ready-issue BUG-123")
636 working_dir: Directory to run the command in
637 issue_id: Optional issue ID for subprocess tracking
639 Returns:
640 CompletedProcess with stdout and stderr
641 """
642 stream_output = self.parallel_config.stream_subprocess_output
644 def stream_callback(line: str, is_stderr: bool) -> None:
645 if stream_output:
646 if is_stderr:
647 print(f" {line}", file=sys.stderr)
648 else:
649 self.logger.info(f" {line}")
651 def on_start(process: subprocess.Popen[str]) -> None:
652 if issue_id:
653 with self._process_lock:
654 self._active_processes[issue_id] = process
656 def on_end(process: subprocess.Popen[str]) -> None:
657 if issue_id:
658 with self._process_lock:
659 self._active_processes.pop(issue_id, None)
661 return _run_claude_base(
662 command=command,
663 timeout=self.parallel_config.timeout_per_issue,
664 working_dir=working_dir,
665 stream_callback=stream_callback if stream_output else None,
666 on_process_start=on_start if issue_id else None,
667 on_process_end=on_end if issue_id else None,
668 idle_timeout=self.parallel_config.idle_timeout_per_issue,
669 )
671 def _run_with_continuation(
672 self,
673 command: str,
674 working_dir: Path,
675 issue_id: str | None = None,
676 max_continuations: int = 3,
677 ) -> subprocess.CompletedProcess[str]:
678 """Run a Claude command with automatic continuation on context handoff.
680 If the command signals CONTEXT_HANDOFF, reads the continuation prompt
681 from the worktree and spawns a fresh Claude session to continue.
683 Args:
684 command: The command to run
685 working_dir: Directory (worktree) to run the command in
686 issue_id: Optional issue ID for subprocess tracking
687 max_continuations: Maximum number of continuation attempts
689 Returns:
690 Combined CompletedProcess with all session outputs
691 """
692 all_stdout: list[str] = []
693 all_stderr: list[str] = []
694 current_command = command
695 continuation_count = 0
696 result: subprocess.CompletedProcess[str] = subprocess.CompletedProcess(
697 args=[], returncode=1, stdout="", stderr=""
698 )
700 while continuation_count <= max_continuations:
701 result = self._run_claude_command(
702 current_command,
703 working_dir,
704 issue_id=issue_id,
705 )
707 all_stdout.append(result.stdout)
708 all_stderr.append(result.stderr)
710 # Check for context handoff signal
711 if detect_context_handoff(result.stdout):
712 self.logger.info(f"[{issue_id}] Detected CONTEXT_HANDOFF signal")
714 # Read continuation prompt from worktree
715 prompt_content = read_continuation_prompt(working_dir)
716 if not prompt_content:
717 self.logger.warning(
718 f"[{issue_id}] Context handoff signaled but no continuation prompt found"
719 )
720 all_stderr.append("Handoff detected but no continuation prompt found")
721 result = subprocess.CompletedProcess(
722 args=result.args, returncode=1, stdout=result.stdout, stderr=result.stderr
723 )
724 break
726 if continuation_count >= max_continuations:
727 self.logger.warning(
728 f"[{issue_id}] Reached max continuations ({max_continuations}), stopping"
729 )
730 break
732 continuation_count += 1
733 self.logger.info(
734 f"[{issue_id}] Starting continuation session #{continuation_count}"
735 )
737 # Re-invoke the original command with --resume flag so the skill
738 # lifecycle (including completion/file-move) runs in the new session.
739 current_command = f"{command} --resume"
740 continue
742 # No handoff signal, we're done
743 break
745 return subprocess.CompletedProcess(
746 args=result.args,
747 returncode=result.returncode,
748 stdout="\n---CONTINUATION---\n".join(all_stdout),
749 stderr="\n---CONTINUATION---\n".join(all_stderr),
750 )
752 def _get_changed_files(self, worktree_path: Path) -> list[str]:
753 """Get list of files changed in the worktree.
755 Args:
756 worktree_path: Path to the worktree
758 Returns:
759 List of changed file paths relative to repo root
760 """
761 result = subprocess.run(
762 ["git", "diff", "--name-only", self.parallel_config.base_branch, "HEAD"],
763 cwd=worktree_path,
764 capture_output=True,
765 text=True,
766 timeout=30,
767 )
769 if result.returncode != 0:
770 return []
772 return [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
774 def _update_branch_base(self, worktree_path: Path, issue_id: str) -> tuple[bool, str]:
775 """Fetch origin/main and rebase worker branch onto it.
777 This ensures the worker branch is based on the latest main before
778 merge coordination, preventing conflicts when main advances during
779 sprint execution (BUG-180).
781 Args:
782 worktree_path: Path to the worker's worktree
783 issue_id: Issue ID for logging
785 Returns:
786 Tuple of (success, error_message)
787 """
788 # Fetch latest base branch from configured remote (fall back to local if fetch fails)
789 base = self.parallel_config.base_branch
790 remote = self.parallel_config.remote_name
791 fetch_result = subprocess.run(
792 ["git", "fetch", remote, base],
793 cwd=worktree_path,
794 capture_output=True,
795 text=True,
796 timeout=60,
797 )
799 rebase_target = f"{remote}/{base}" if fetch_result.returncode == 0 else base
801 # Rebase current branch onto base (remote or local fallback)
802 rebase_result = subprocess.run(
803 ["git", "rebase", rebase_target],
804 cwd=worktree_path,
805 capture_output=True,
806 text=True,
807 timeout=120,
808 )
810 if rebase_result.returncode != 0:
811 # Abort the failed rebase
812 subprocess.run(
813 ["git", "rebase", "--abort"],
814 cwd=worktree_path,
815 capture_output=True,
816 timeout=10,
817 )
818 return False, f"Failed to rebase onto {rebase_target}: {rebase_result.stderr}"
820 self.logger.info(f"[{issue_id}] Rebased branch onto {rebase_target}")
821 return True, ""
823 def _verify_work_was_done(
824 self, changed_files: list[str], issue_id: str, issue_filename: str = ""
825 ) -> tuple[bool, str]:
826 """Verify that actual implementation work was done.
828 Uses the shared verify_work_was_done() function to check that changed
829 files include meaningful work, not just issue files or other artifacts.
831 Args:
832 changed_files: List of files changed during processing
833 issue_id: The issue ID being processed (unused, kept for compatibility)
834 issue_filename: Full issue filename (unused, kept for compatibility)
836 Returns:
837 Tuple of (success, error_message)
838 """
839 if not changed_files:
840 return False, "No files were changed during implementation"
842 # Check if code changes are required
843 if not self.parallel_config.require_code_changes:
844 return True, ""
846 # Use shared verification function
847 if verify_work_was_done(self.logger, changed_files):
848 return True, ""
850 # Generate descriptive error with actual excluded files
851 excluded_files = [
852 f
853 for f in changed_files
854 if f and any(f.startswith(excl) for excl in EXCLUDED_DIRECTORIES)
855 ]
856 if excluded_files:
857 files_preview = ", ".join(excluded_files[:5])
858 if len(excluded_files) > 5:
859 files_preview += f" (+{len(excluded_files) - 5} more)"
860 return False, f"Only excluded files modified: {files_preview}"
861 return False, "Only excluded files modified (e.g., .issues/, thoughts/)"
863 def _has_other_issue_id(self, file_lower: str, current_issue_id_lower: str) -> bool:
864 """Check if file contains a different issue ID than the current worker's.
866 This prevents cross-worker contamination where worker A detects worker B's
867 leaked file. When multiple workers run in parallel, their leaked files may
868 both appear in the main repo. Each worker should only clean up its own leaks.
870 Args:
871 file_lower: Lowercase file path to check
872 current_issue_id_lower: Lowercase issue ID of the current worker
874 Returns:
875 True if the file contains a different issue ID (belongs to another worker),
876 False if the file contains the current issue ID or no recognizable issue ID
877 """
878 # Pattern matches common issue ID formats: BUG-123, ENH-456, FEAT-789
879 # Use non-capturing group (?:...) so findall returns full match, not group
880 matches = re.findall(r"(?:bug|enh|feat)-\d+", file_lower)
882 if not matches:
883 # No issue ID found - file doesn't belong to any specific worker
884 return False
886 # Check if any of the found issue IDs match the current worker
887 for match in matches:
888 if match == current_issue_id_lower:
889 return False # File belongs to current worker
891 # File has issue ID(s) but none match current worker - belongs to another worker
892 return True
894 def _detect_main_repo_leaks(self, issue_id: str, baseline_status: set[str]) -> list[str]:
895 """Detect files incorrectly written to main repo instead of worktree.
897 Claude Code may write files to the main repository instead of the
898 worktree due to project root detection issues (see GitHub #8771).
899 This method detects such leaks by comparing main repo status before
900 and after worker execution.
902 Args:
903 issue_id: ID of the issue being processed (for pattern matching)
904 baseline_status: Set of file paths from git status before worker started
906 Returns:
907 List of file paths that were leaked to main repo
908 """
909 # Get current status of main repo
910 result = self._git_lock.run(
911 ["status", "--porcelain"],
912 cwd=self.repo_path,
913 timeout=30,
914 )
916 if result.returncode != 0:
917 return []
919 current_files: set[str] = set()
920 for line in result.stdout.strip().split("\n"):
921 if not line or len(line) < 3:
922 continue
923 # Extract file path (after status codes and space)
924 file_path = line[3:].strip()
925 # Handle renamed files (old -> new)
926 if " -> " in file_path:
927 file_path = file_path.split(" -> ")[-1]
928 current_files.add(file_path)
930 # Find new files that appeared during worker execution
931 new_files = current_files - baseline_status
933 # Filter to files likely related to this issue
934 issue_id_lower = issue_id.lower()
935 leaked_files: list[str] = []
937 # Build source prefix list: start with common fallbacks, then add configured dirs
938 source_prefixes = ["backend/", "src/", "lib/", "tests/"]
939 for dir_path in [self.br_config.project.src_dir, self.br_config.project.test_dir]:
940 if dir_path:
941 normalized = dir_path.rstrip("/") + "/"
942 if normalized not in source_prefixes:
943 source_prefixes.append(normalized)
945 for file_path in new_files:
946 # Skip state file (managed by orchestrator)
947 if file_path.endswith(".parallel-manage-state.json"):
948 continue
949 # Skip .gitignore (may be modified by ll-parallel)
950 if file_path == ".gitignore":
951 continue
953 # Check if file is related to this issue
954 file_lower = file_path.lower()
955 if issue_id_lower in file_lower:
956 leaked_files.append(file_path)
957 # Also catch source files that shouldn't be modified in main
958 elif file_path.startswith(tuple(source_prefixes)):
959 leaked_files.append(file_path)
960 # Catch thoughts/plans files
961 elif file_path.startswith("thoughts/"):
962 leaked_files.append(file_path)
963 # Catch issue files in any issue directory variant
964 # Handles both .issues/ (with dot) and issues/ (without dot)
965 # Only include files without a different issue ID - files WITH other issue IDs
966 # belong to other workers running in parallel (cross-worker contamination)
967 elif file_path.startswith((".issues/", "issues/")):
968 if not self._has_other_issue_id(file_lower, issue_id_lower):
969 leaked_files.append(file_path)
971 return leaked_files
973 def _cleanup_leaked_files(self, leaked_files: list[str]) -> int:
974 """Discard leaked files from main repo working directory.
976 Claude Code sometimes writes files to the main repo instead of the
977 worktree. These files cause stash conflicts during merge operations.
978 Since the actual work is preserved in the worktree branch, we can
979 safely discard these leaked changes from the main repo.
981 Args:
982 leaked_files: List of file paths leaked to main repo
984 Returns:
985 Number of files successfully cleaned up
986 """
987 if not leaked_files:
988 return 0
990 cleaned = 0
992 # Get status to determine which files are tracked vs untracked
993 status_result = self._git_lock.run(
994 ["status", "--porcelain", "--"] + leaked_files,
995 cwd=self.repo_path,
996 timeout=30,
997 )
999 tracked_files: list[str] = []
1000 untracked_files: list[str] = []
1002 for line in status_result.stdout.splitlines():
1003 if not line or len(line) < 3:
1004 continue
1005 status_code = line[:2]
1006 file_path = line[3:].split(" -> ")[-1].strip()
1008 if status_code.startswith("?"):
1009 # Untracked file - need to delete
1010 untracked_files.append(file_path)
1011 else:
1012 # Tracked file - can use git checkout to discard
1013 tracked_files.append(file_path)
1015 # Discard changes to tracked files
1016 if tracked_files:
1017 checkout_result = self._git_lock.run(
1018 ["checkout", "--"] + tracked_files,
1019 cwd=self.repo_path,
1020 timeout=30,
1021 )
1022 if checkout_result.returncode == 0:
1023 cleaned += len(tracked_files)
1024 else:
1025 self.logger.warning(
1026 f"Failed to discard tracked leaked files: {checkout_result.stderr}"
1027 )
1029 # Delete untracked files
1030 for file_path in untracked_files:
1031 full_path = self.repo_path / file_path
1032 try:
1033 if full_path.exists():
1034 full_path.unlink()
1035 cleaned += 1
1036 except OSError as e:
1037 self.logger.warning(f"Failed to delete leaked file {file_path}: {e}")
1039 # Fallback: directly delete files not reported by git status
1040 # This handles gitignored files that git status --porcelain doesn't show
1041 accounted_files = set(tracked_files + untracked_files)
1042 for file_path in leaked_files:
1043 if file_path not in accounted_files:
1044 full_path = self.repo_path / file_path
1045 if full_path.exists():
1046 try:
1047 full_path.unlink()
1048 cleaned += 1
1049 self.logger.info(f"Deleted gitignored leaked file: {file_path}")
1050 except OSError as e:
1051 self.logger.warning(
1052 f"Failed to delete gitignored leaked file {file_path}: {e}"
1053 )
1054 else:
1055 self.logger.debug(f"Leaked file not found (may have been moved): {file_path}")
1057 if cleaned > 0:
1058 self.logger.info(f"Cleaned up {cleaned} leaked file(s) from main repo")
1060 return cleaned
1062 def _get_main_repo_baseline(self) -> set[str]:
1063 """Get baseline of modified/untracked files in main repo.
1065 Returns:
1066 Set of file paths currently showing in git status
1067 """
1068 result = self._git_lock.run(
1069 ["status", "--porcelain"],
1070 cwd=self.repo_path,
1071 timeout=30,
1072 )
1074 if result.returncode != 0:
1075 return set()
1077 files: set[str] = set()
1078 for line in result.stdout.strip().split("\n"):
1079 if not line or len(line) < 3:
1080 continue
1081 file_path = line[3:].strip()
1082 if " -> " in file_path:
1083 file_path = file_path.split(" -> ")[-1]
1084 files.add(file_path)
1086 return files
1088 def _get_main_head_sha(self) -> str:
1089 """Get the current HEAD SHA of the main repo.
1091 Returns:
1092 HEAD SHA string, or empty string if unavailable
1093 """
1094 result = self._git_lock.run(
1095 ["rev-parse", "HEAD"],
1096 cwd=self.repo_path,
1097 timeout=10,
1098 )
1099 if result.returncode == 0:
1100 return result.stdout.strip()
1101 return ""
1103 def _detect_committed_leaks(self, baseline_head_sha: str) -> list[str]:
1104 """Detect commits made directly to main repo during worker execution.
1106 When Claude commits to the main repo instead of the worktree branch,
1107 the commits appear on main's history but the worktree has no changes.
1108 This method detects such leaked commits by comparing main's HEAD SHA
1109 before and after worker execution.
1111 Args:
1112 baseline_head_sha: HEAD SHA captured before worker started
1114 Returns:
1115 List of commit SHAs committed to main during worker execution,
1116 newest first. Empty list if no committed leaks detected.
1117 """
1118 if not baseline_head_sha:
1119 return []
1121 current_sha = self._get_main_head_sha()
1122 if not current_sha or current_sha == baseline_head_sha:
1123 return []
1125 # Get list of new commits on main since baseline
1126 result = self._git_lock.run(
1127 ["log", "--format=%H", f"{baseline_head_sha}..HEAD"],
1128 cwd=self.repo_path,
1129 timeout=30,
1130 )
1131 if result.returncode != 0:
1132 return []
1134 commits = [sha.strip() for sha in result.stdout.strip().split("\n") if sha.strip()]
1135 return commits
1137 def _recover_committed_leaks(
1138 self,
1139 leaked_commits: list[str],
1140 worktree_path: Path,
1141 baseline_head_sha: str,
1142 issue_id: str,
1143 ) -> bool:
1144 """Attempt to recover committed leaks by cherry-picking to worktree.
1146 When Claude commits directly to main instead of the worktree branch,
1147 we attempt to:
1148 1. Cherry-pick the leaked commits onto the worktree branch
1149 2. Reset main back to the baseline SHA (if safe to do so)
1151 This preserves the implementation work in the worktree while
1152 cleaning up the incorrect commits on main.
1154 Args:
1155 leaked_commits: Commit SHAs that leaked to main (newest first)
1156 worktree_path: Path to the worker's worktree
1157 baseline_head_sha: Main HEAD SHA before worker started
1158 issue_id: Issue ID for logging
1160 Returns:
1161 True if cherry-pick succeeded (main reset is attempted but
1162 not required for a True return value)
1163 """
1164 self.logger.info(
1165 f"[{issue_id}] Attempting recovery: cherry-picking {len(leaked_commits)} "
1166 f"commit(s) to worktree"
1167 )
1169 # Cherry-pick in chronological order (oldest first = reverse of log output)
1170 for sha in reversed(leaked_commits):
1171 result = subprocess.run(
1172 ["git", "cherry-pick", sha],
1173 cwd=worktree_path,
1174 capture_output=True,
1175 text=True,
1176 timeout=60,
1177 )
1178 if result.returncode != 0:
1179 subprocess.run(
1180 ["git", "cherry-pick", "--abort"],
1181 cwd=worktree_path,
1182 capture_output=True,
1183 timeout=10,
1184 )
1185 self.logger.warning(
1186 f"[{issue_id}] Cherry-pick of {sha[:8]} failed: {result.stderr.strip()}"
1187 )
1188 return False
1190 # Attempt to reset main to baseline (only if main hasn't advanced further)
1191 current_main_sha = self._get_main_head_sha()
1192 most_recent_leaked = leaked_commits[0] # Newest first
1193 if current_main_sha == most_recent_leaked:
1194 reset_result = self._git_lock.run(
1195 ["reset", "--hard", baseline_head_sha],
1196 cwd=self.repo_path,
1197 timeout=30,
1198 )
1199 if reset_result.returncode == 0:
1200 self.logger.info(f"[{issue_id}] Reset main to baseline {baseline_head_sha[:8]}")
1201 else:
1202 self.logger.warning(
1203 f"[{issue_id}] Cherry-pick succeeded but failed to reset main: "
1204 f"{reset_result.stderr.strip()}"
1205 )
1206 else:
1207 # main has advanced past the leaked commits — attempt surgical rebase
1208 # to excise only the leaked commits while preserving subsequent work
1209 self.logger.info(
1210 f"[{issue_id}] Main has advanced beyond leaked commits "
1211 f"({current_main_sha[:8]} != {most_recent_leaked[:8]}) — "
1212 f"attempting surgical rebase to excise leaked commits"
1213 )
1214 rebase_result = self._git_lock.run(
1215 ["rebase", "--onto", baseline_head_sha, most_recent_leaked],
1216 cwd=self.repo_path,
1217 timeout=60,
1218 )
1219 if rebase_result.returncode == 0:
1220 self.logger.info(f"[{issue_id}] Surgically removed leaked commits via rebase")
1221 else:
1222 self._git_lock.run(
1223 ["rebase", "--abort"],
1224 cwd=self.repo_path,
1225 timeout=10,
1226 )
1227 self.logger.warning(
1228 f"[{issue_id}] Surgical rebase failed — manual cleanup required: "
1229 f"{rebase_result.stderr.strip()}"
1230 )
1232 self.logger.info(
1233 f"[{issue_id}] Recovered {len(leaked_commits)} commit(s): "
1234 f"cherry-picked to worktree branch"
1235 )
1236 return True
1238 @property
1239 def active_count(self) -> int:
1240 """Number of currently active workers.
1242 Includes both workers with running futures AND workers whose futures
1243 are done but callbacks haven't completed yet.
1244 """
1245 with self._process_lock:
1246 running_futures = sum(1 for f in self._active_workers.values() if not f.done())
1247 with self._callback_lock:
1248 pending_callback_count = len(self._pending_callbacks)
1249 return running_futures + pending_callback_count
1251 def set_worker_stage(self, issue_id: str, stage: WorkerStage) -> None:
1252 """Update the stage of a worker.
1254 Args:
1255 issue_id: Issue ID being processed
1256 stage: New stage value
1257 """
1258 with self._process_lock:
1259 self._worker_stages[issue_id] = stage
1261 def get_worker_stage(self, issue_id: str) -> WorkerStage | None:
1262 """Get the current stage of a worker.
1264 Args:
1265 issue_id: Issue ID being processed
1267 Returns:
1268 Current stage, or None if issue not being tracked
1269 """
1270 with self._process_lock:
1271 return self._worker_stages.get(issue_id)
1273 def get_active_stages(self) -> dict[str, WorkerStage]:
1274 """Get all active worker stages.
1276 Returns:
1277 Dictionary mapping issue_id to current stage for active workers
1278 """
1279 with self._process_lock:
1280 # Only return workers that are actually active
1281 active_ids = set(self._active_workers.keys())
1282 return {
1283 issue_id: stage
1284 for issue_id, stage in self._worker_stages.items()
1285 if issue_id in active_ids
1286 }
1288 def remove_worker_stage(self, issue_id: str) -> None:
1289 """Remove a worker from stage tracking.
1291 Args:
1292 issue_id: Issue ID to remove
1293 """
1294 with self._process_lock:
1295 self._worker_stages.pop(issue_id, None)
1297 def cleanup_all_worktrees(self) -> None:
1298 """Clean up all worker worktrees."""
1299 worktree_base = self.repo_path / self.parallel_config.worktree_base
1300 if not worktree_base.exists():
1301 return
1303 for worktree_dir in worktree_base.iterdir():
1304 if worktree_dir.is_dir() and worktree_dir.name.startswith("worker-"):
1305 self._cleanup_worktree(worktree_dir)
1307 self.logger.info("Cleaned up all worker worktrees")