Coverage for little_loops / parallel / worker_pool.py: 10%
503 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -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 assemble_guillotine_prompt,
27 detect_context_handoff,
28 read_continuation_prompt,
29 read_sentinel,
30 write_sentinel,
31)
32from little_loops.subprocess_utils import (
33 run_claude_command as _run_claude_base,
34)
35from little_loops.work_verification import EXCLUDED_DIRECTORIES, verify_work_was_done
37if TYPE_CHECKING:
38 from little_loops.config import BRConfig
39 from little_loops.issue_parser import IssueInfo
40 from little_loops.logger import Logger
43class WorkerPool:
44 """Thread pool for processing issues in isolated git worktrees.
46 Each worker:
47 1. Creates a dedicated git worktree and branch
48 2. Runs issue validation and implementation via Claude CLI
49 3. Commits changes locally
50 4. Returns results for merge coordination
52 Example:
53 >>> pool = WorkerPool(parallel_config, br_config, logger)
54 >>> pool.start()
55 >>> future = pool.submit(issue_info)
56 >>> result = future.result() # WorkerResult
57 >>> pool.shutdown()
58 """
60 def __init__(
61 self,
62 parallel_config: ParallelConfig,
63 br_config: BRConfig,
64 logger: Logger,
65 repo_path: Path | None = None,
66 git_lock: GitLock | None = None,
67 ) -> None:
68 """Initialize the worker pool.
70 Args:
71 parallel_config: Parallel processing configuration
72 br_config: Project configuration (for category actions)
73 logger: Logger for worker output
74 repo_path: Path to the git repository (default: current directory)
75 git_lock: Shared lock for git operations (created if not provided)
76 """
77 self.parallel_config = parallel_config
78 self.br_config = br_config
79 self.logger = logger
80 self.repo_path = repo_path or Path.cwd()
81 self._git_lock = git_lock or GitLock(logger)
82 self._executor: ThreadPoolExecutor | None = None
83 self._active_workers: dict[str, Future[WorkerResult]] = {}
84 # Track active subprocesses for forceful termination on shutdown
85 self._active_processes: dict[str, subprocess.Popen[str]] = {}
86 # Track active worktree paths to prevent cleanup while in use (BUG-142)
87 self._active_worktrees: set[Path] = set()
88 self._process_lock = threading.Lock()
89 # Track callbacks currently executing
90 self._pending_callbacks: set[str] = set()
91 self._callback_lock = threading.Lock()
92 # Shutdown tracking for interrupted worker detection (ENH-036)
93 self._shutdown_requested = False
94 self._terminated_during_shutdown: set[str] = set()
95 # Track worker processing stages for progress visibility (ENH-262)
96 self._worker_stages: dict[str, WorkerStage] = {}
98 def start(self) -> None:
99 """Start the worker pool."""
100 if self._executor is not None:
101 return
103 # Ensure worktree base directory exists
104 worktree_base = self.repo_path / self.parallel_config.worktree_base
105 worktree_base.mkdir(parents=True, exist_ok=True)
107 self._executor = ThreadPoolExecutor(
108 max_workers=self.parallel_config.max_workers,
109 thread_name_prefix="issue-worker",
110 )
111 self.logger.info(f"Worker pool started with {self.parallel_config.max_workers} workers")
113 def shutdown(self, wait: bool = True) -> None:
114 """Shutdown the worker pool.
116 Args:
117 wait: Whether to wait for pending tasks to complete
118 """
119 if self._executor is None:
120 return
122 self.logger.info("Shutting down worker pool...")
124 # First, terminate all active subprocesses to unblock worker threads
125 if not wait:
126 self.terminate_all_processes()
128 self._executor.shutdown(wait=wait)
129 self._executor = None
131 def set_shutdown_requested(self, value: bool = True) -> None:
132 """Set the shutdown flag.
134 Called by orchestrator during shutdown to enable tracking of
135 workers that are terminated due to shutdown vs. actual failures.
136 """
137 self._shutdown_requested = value
139 def terminate_all_processes(self) -> None:
140 """Forcefully terminate all active subprocesses.
142 Called when we need to abort workers immediately,
143 such as on timeout or shutdown.
144 """
145 with self._process_lock:
146 for issue_id, process in list(self._active_processes.items()):
147 if process.poll() is None: # Still running
148 self.logger.warning(
149 f"Terminating subprocess for {issue_id} (PID {process.pid})"
150 )
151 # Track issues terminated during shutdown for interrupted detection (ENH-036)
152 if self._shutdown_requested:
153 self._terminated_during_shutdown.add(issue_id)
154 try:
155 # Send SIGTERM first for graceful termination
156 process.terminate()
157 try:
158 process.wait(timeout=5)
159 except subprocess.TimeoutExpired:
160 # Force kill if SIGTERM didn't work
161 self.logger.warning(f"Force killing {issue_id} (PID {process.pid})")
162 process.kill()
163 process.wait(timeout=2)
164 except Exception as e:
165 self.logger.error(f"Failed to terminate {issue_id}: {e}")
166 self._active_processes.clear()
168 def submit(
169 self,
170 issue: IssueInfo,
171 on_complete: Callable[[WorkerResult], None] | None = None,
172 ) -> Future[WorkerResult]:
173 """Submit an issue for processing.
175 Args:
176 issue: Issue to process
177 on_complete: Optional callback when processing completes
179 Returns:
180 Future that will contain the WorkerResult
181 """
182 if self._executor is None:
183 raise RuntimeError("Worker pool not started")
185 future = self._executor.submit(self._process_issue, issue)
186 with self._process_lock:
187 self._active_workers[issue.issue_id] = future
189 if on_complete:
190 future.add_done_callback(
191 lambda f: self._handle_completion(f, on_complete, issue.issue_id)
192 )
194 return future
196 def _handle_completion(
197 self,
198 future: Future[WorkerResult],
199 callback: Callable[[WorkerResult], None],
200 issue_id: str,
201 ) -> None:
202 """Handle worker completion and invoke callback."""
203 with self._callback_lock:
204 self._pending_callbacks.add(issue_id)
205 try:
206 try:
207 result = future.result()
208 except Exception as e:
209 self.logger.error(f"Worker future failed for {issue_id}: {e}")
210 result = WorkerResult(
211 issue_id=issue_id,
212 success=False,
213 branch_name="",
214 worktree_path=Path(),
215 error=f"Worker future failed: {e}",
216 )
217 # Set final stage based on result (ENH-262)
218 if result.success:
219 self.set_worker_stage(issue_id, WorkerStage.COMPLETED)
220 elif result.interrupted:
221 self.set_worker_stage(issue_id, WorkerStage.INTERRUPTED)
222 else:
223 self.set_worker_stage(issue_id, WorkerStage.FAILED)
224 try:
225 callback(result)
226 except Exception as e:
227 self.logger.error(f"Worker completion callback failed for {issue_id}: {e}")
228 finally:
229 with self._callback_lock:
230 self._pending_callbacks.discard(issue_id)
232 def _process_issue(self, issue: IssueInfo) -> WorkerResult:
233 """Process a single issue in an isolated worktree.
235 Args:
236 issue: Issue to process
238 Returns:
239 WorkerResult with processing outcome
240 """
241 start_time = time.time()
242 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
243 if self.parallel_config.use_feature_branches:
244 from little_loops.issue_parser import slugify
246 branch_name = f"feature/{issue.issue_id.lower()}-{slugify(issue.title)}"
247 else:
248 branch_name = f"parallel/{issue.issue_id.lower()}-{timestamp}"
249 worktree_path = (
250 self.repo_path
251 / self.parallel_config.worktree_base
252 / f"worker-{issue.issue_id.lower()}-{timestamp}"
253 )
255 # Set initial stage for progress tracking (ENH-262)
256 self.set_worker_stage(issue.issue_id, WorkerStage.SETUP)
258 # Capture baseline of main repo status before worker starts
259 # Used to detect files incorrectly written to main repo
260 baseline_status = self._get_main_repo_baseline()
261 # Capture main HEAD SHA before worker starts to detect committed leaks
262 baseline_head_sha = self._get_main_head_sha()
264 try:
265 # Step 1: Create worktree with new branch
266 self._setup_worktree(worktree_path, branch_name)
268 # Register worktree as active to prevent cleanup while in use (BUG-142)
269 with self._process_lock:
270 self._active_worktrees.add(worktree_path)
272 # Update stage for progress tracking (ENH-262)
273 self.set_worker_stage(issue.issue_id, WorkerStage.VALIDATING)
275 # Step 2: Run ready-issue validation
276 ready_cmd = self.parallel_config.get_ready_command(issue.issue_id)
277 ready_result = self._run_claude_command(
278 ready_cmd,
279 worktree_path,
280 issue_id=issue.issue_id,
281 )
283 # Check if worker was terminated during shutdown (ENH-036)
284 if issue.issue_id in self._terminated_during_shutdown:
285 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
286 return WorkerResult(
287 issue_id=issue.issue_id,
288 success=False,
289 interrupted=True,
290 branch_name=branch_name,
291 worktree_path=worktree_path,
292 duration=time.time() - start_time,
293 error="Interrupted during shutdown",
294 stdout=ready_result.stdout,
295 stderr=ready_result.stderr,
296 )
298 if ready_result.returncode != 0:
299 err_detail = ready_result.stderr or (ready_result.stdout or "")[:500]
300 return WorkerResult(
301 issue_id=issue.issue_id,
302 success=False,
303 branch_name=branch_name,
304 worktree_path=worktree_path,
305 duration=time.time() - start_time,
306 error=f"ready-issue failed: {err_detail}",
307 stdout=ready_result.stdout,
308 stderr=ready_result.stderr,
309 )
311 # Step 3: Parse ready-issue output and check verdict
312 ready_parsed = parse_ready_issue_output(ready_result.stdout)
314 # Handle CLOSE verdict - issue should not be implemented
315 if ready_parsed.get("should_close"):
316 return WorkerResult(
317 issue_id=issue.issue_id,
318 success=True, # Closure is a valid outcome
319 branch_name=branch_name,
320 worktree_path=worktree_path,
321 duration=time.time() - start_time,
322 should_close=True,
323 close_reason=ready_parsed.get("close_reason"),
324 close_status=ready_parsed.get("close_status"),
325 stdout=ready_result.stdout,
326 stderr=ready_result.stderr,
327 )
329 # Handle BLOCKED verdict - issue has open dependencies
330 if ready_parsed.get("is_blocked"):
331 return WorkerResult(
332 issue_id=issue.issue_id,
333 success=False,
334 was_blocked=True,
335 branch_name=branch_name,
336 worktree_path=worktree_path,
337 duration=time.time() - start_time,
338 error="ready-issue verdict: BLOCKED - open dependency detected",
339 stdout=ready_result.stdout,
340 stderr=ready_result.stderr,
341 )
343 # Handle NOT_READY verdict
344 if not ready_parsed["is_ready"]:
345 concerns = ready_parsed.get("concerns", [])
346 if concerns:
347 concern_msg = "; ".join(concerns)
348 elif ready_parsed["verdict"] == "UNKNOWN":
349 # For UNKNOWN verdicts, show a snippet of output for debugging
350 raw_out = (ready_result.stdout or "")[:200].strip()
351 concern_msg = (
352 f"Could not parse verdict. Output: {raw_out}..."
353 if raw_out
354 else "No output from ready-issue"
355 )
356 else:
357 concern_msg = "Issue not ready"
358 return WorkerResult(
359 issue_id=issue.issue_id,
360 success=False,
361 branch_name=branch_name,
362 worktree_path=worktree_path,
363 duration=time.time() - start_time,
364 error=f"ready-issue verdict: {ready_parsed['verdict']} - {concern_msg}",
365 stdout=ready_result.stdout,
366 stderr=ready_result.stderr,
367 )
369 # Track if issue was corrected (corrections stay in worktree)
370 was_corrected = ready_parsed.get("was_corrected", False)
371 corrections = ready_parsed.get("corrections", [])
373 # Update stage for progress tracking (ENH-262)
374 self.set_worker_stage(issue.issue_id, WorkerStage.IMPLEMENTING)
376 # Decision gate: invoke decide-issue when the issue requires a decision
377 if issue.decision_needed is True:
378 decide_cmd = self.parallel_config.get_decide_command(issue.issue_id)
379 decide_result = self._run_claude_command(
380 decide_cmd, worktree_path, issue_id=issue.issue_id
381 )
382 if decide_result.returncode != 0:
383 self.logger.warning(
384 f"[{issue.issue_id}] decide-issue command failed, "
385 "continuing to implementation anyway..."
386 )
388 # Step 4: Get action from BRConfig
389 action = self.br_config.get_category_action(issue.issue_type)
391 # Step 5: Run manage-issue implementation (with continuation support)
392 manage_cmd = self.parallel_config.get_manage_command(
393 issue.issue_type, action, issue.issue_id
394 )
395 manage_result = self._run_with_continuation(
396 manage_cmd,
397 worktree_path,
398 issue_id=issue.issue_id,
399 )
401 # Update stage for progress tracking (ENH-262)
402 self.set_worker_stage(issue.issue_id, WorkerStage.VERIFYING)
404 # Check if worker was terminated during shutdown (ENH-036)
405 if issue.issue_id in self._terminated_during_shutdown:
406 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
407 return WorkerResult(
408 issue_id=issue.issue_id,
409 success=False,
410 interrupted=True,
411 branch_name=branch_name,
412 worktree_path=worktree_path,
413 duration=time.time() - start_time,
414 error="Interrupted during shutdown",
415 stdout=manage_result.stdout,
416 stderr=manage_result.stderr,
417 )
419 # Step 6: Get list of changed files in worktree
420 changed_files = self._get_changed_files(worktree_path)
422 # Step 8: Detect files leaked to main repo instead of worktree (unstaged)
423 leaked_files = self._detect_main_repo_leaks(issue.issue_id, baseline_status)
424 if leaked_files:
425 self.logger.warning(
426 f"{issue.issue_id} leaked {len(leaked_files)} file(s) to main repo: "
427 f"{leaked_files}"
428 )
429 # Clean up leaked files to prevent stash conflicts during merge.
430 # The actual work is preserved in the worktree branch.
431 self._cleanup_leaked_files(leaked_files)
433 # Step 8b: Detect commits made directly to main instead of worktree branch.
434 # If Claude committed to main (not the worktree), worktree will have no diff,
435 # causing work verification to fail. Attempt to recover by cherry-picking
436 # the leaked commits to the worktree and resetting main. (BUG-580)
437 committed_leaks = self._detect_committed_leaks(baseline_head_sha)
438 if committed_leaks:
439 self.logger.warning(
440 f"{issue.issue_id} committed {len(committed_leaks)} commit(s) directly "
441 f"to main instead of worktree: {[sha[:8] for sha in committed_leaks]}"
442 )
443 if not changed_files:
444 recovered = self._recover_committed_leaks(
445 committed_leaks, worktree_path, baseline_head_sha, issue.issue_id
446 )
447 if recovered:
448 changed_files = self._get_changed_files(worktree_path)
450 # Step 7: Verify actual work was done (after potential committed-leak recovery)
451 # Pass full filename for better doc-only keyword matching
452 issue_filename = issue.path.stem if issue.path else ""
453 work_verified, verification_error = self._verify_work_was_done(
454 changed_files, issue.issue_id, issue_filename
455 )
457 if manage_result.returncode != 0:
458 err_detail = manage_result.stderr or (manage_result.stdout or "")[:500]
459 return WorkerResult(
460 issue_id=issue.issue_id,
461 success=False,
462 branch_name=branch_name,
463 worktree_path=worktree_path,
464 changed_files=changed_files,
465 leaked_files=leaked_files,
466 duration=time.time() - start_time,
467 error=f"manage-issue failed: {err_detail}",
468 stdout=manage_result.stdout,
469 stderr=manage_result.stderr,
470 )
472 if not work_verified:
473 return WorkerResult(
474 issue_id=issue.issue_id,
475 success=False,
476 branch_name=branch_name,
477 worktree_path=worktree_path,
478 changed_files=changed_files,
479 leaked_files=leaked_files,
480 duration=time.time() - start_time,
481 error=verification_error,
482 stdout=manage_result.stdout,
483 stderr=manage_result.stderr,
484 )
486 # Step 9: Update branch base before merge (BUG-180)
487 # Fetch origin/main and rebase to ensure branch is based on latest main
488 base_updated, base_error = self._update_branch_base(worktree_path, issue.issue_id)
490 # Update stage for progress tracking (ENH-262)
491 self.set_worker_stage(issue.issue_id, WorkerStage.MERGING)
493 if not base_updated:
494 return WorkerResult(
495 issue_id=issue.issue_id,
496 success=False,
497 branch_name=branch_name,
498 worktree_path=worktree_path,
499 changed_files=changed_files,
500 leaked_files=leaked_files,
501 duration=time.time() - start_time,
502 error=base_error,
503 stdout=manage_result.stdout,
504 stderr=manage_result.stderr,
505 )
507 return WorkerResult(
508 issue_id=issue.issue_id,
509 success=True,
510 branch_name=branch_name,
511 worktree_path=worktree_path,
512 changed_files=changed_files,
513 leaked_files=leaked_files,
514 duration=time.time() - start_time,
515 error=None,
516 stdout=manage_result.stdout,
517 stderr=manage_result.stderr,
518 was_corrected=was_corrected,
519 corrections=corrections,
520 )
522 except Exception as e:
523 return WorkerResult(
524 issue_id=issue.issue_id,
525 success=False,
526 branch_name=branch_name,
527 worktree_path=worktree_path,
528 duration=time.time() - start_time,
529 error=str(e),
530 )
531 finally:
532 # Unregister worktree as no longer active (BUG-142)
533 with self._process_lock:
534 self._active_worktrees.discard(worktree_path)
536 def _setup_worktree(self, worktree_path: Path, branch_name: str) -> None:
537 """Create a git worktree with a new branch.
539 Args:
540 worktree_path: Path for the new worktree
541 branch_name: Name of the new branch
542 """
543 from little_loops.worktree_utils import setup_worktree
545 setup_worktree(
546 repo_path=self.repo_path,
547 worktree_path=worktree_path,
548 branch_name=branch_name,
549 copy_files=self.parallel_config.worktree_copy_files,
550 logger=self.logger,
551 git_lock=self._git_lock,
552 )
554 # Verify model if --show-model flag is set (requires API call)
555 if self.parallel_config.show_model:
556 model = self._detect_worktree_model_via_api(worktree_path)
557 if model:
558 self.logger.info(f" Using model: {model}")
559 else:
560 self.logger.warning(" Could not detect Claude CLI model")
562 def _detect_worktree_model_via_api(self, worktree_path: Path) -> str | None:
563 """Detect the model Claude will use by making an API call.
565 Runs a minimal Claude command with JSON output and parses the modelUsage
566 field to verify settings.local.json is being respected.
568 Args:
569 worktree_path: Path to the worktree to test
571 Returns:
572 Model name (e.g., "claude-sonnet-4-20250514") or None if unable to detect
573 """
574 try:
575 # Set environment to keep Claude in the project working directory (BUG-007)
576 # This ensures the first Claude CLI invocation in the worktree has the same
577 # project root behavior as subsequent invocations via run_claude_command()
578 env = os.environ.copy()
579 env["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1"
581 # Use a minimal prompt that requires an API call to get modelUsage
582 result = subprocess.run(
583 [
584 "claude",
585 "-p",
586 "reply with just 'ok'",
587 "--output-format",
588 "json",
589 ],
590 cwd=worktree_path,
591 capture_output=True,
592 text=True,
593 timeout=30,
594 env=env,
595 )
596 if result.returncode == 0 and result.stdout.strip():
597 data: dict[str, Any] = json.loads(result.stdout.strip())
598 model_usage: dict[str, Any] = data.get("modelUsage", {})
599 # Return the first (primary) model from modelUsage
600 if model_usage:
601 return cast(str, next(iter(model_usage.keys())))
602 except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError):
603 pass
604 return None
606 def _cleanup_worktree(self, worktree_path: Path) -> None:
607 """Remove a git worktree and its associated branch.
609 Args:
610 worktree_path: Path to the worktree to remove
611 """
612 if not worktree_path.exists():
613 return
615 # Skip cleanup if worktree is actively in use by a running worker (BUG-142)
616 with self._process_lock:
617 if worktree_path in self._active_worktrees:
618 self.logger.warning(
619 f"Skipping cleanup of {worktree_path.name}: worktree is in active use"
620 )
621 return
623 # Only delete branches with the parallel/ prefix (legacy behavior for ll-parallel)
624 branch_result = subprocess.run(
625 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
626 cwd=worktree_path,
627 capture_output=True,
628 text=True,
629 )
630 branch_name = branch_result.stdout.strip() if branch_result.returncode == 0 else None
631 delete_branch = branch_name is not None and branch_name.startswith("parallel/")
633 from little_loops.worktree_utils import cleanup_worktree
635 cleanup_worktree(
636 worktree_path=worktree_path,
637 repo_path=self.repo_path,
638 logger=self.logger,
639 git_lock=self._git_lock,
640 delete_branch=delete_branch,
641 )
643 def _run_claude_command(
644 self,
645 command: str,
646 working_dir: Path,
647 issue_id: str | None = None,
648 on_usage: Callable[[int, int], None] | None = None,
649 resume_session: bool = False,
650 ) -> subprocess.CompletedProcess[str]:
651 """Run a Claude CLI command with real-time output streaming.
653 Args:
654 command: The command to run (e.g., "/ll:ready-issue BUG-123")
655 working_dir: Directory to run the command in
656 issue_id: Optional issue ID for subprocess tracking
657 on_usage: Optional usage callback for token tracking
658 resume_session: If True, passes --continue to the Claude CLI
660 Returns:
661 CompletedProcess with stdout and stderr
662 """
663 stream_output = self.parallel_config.stream_subprocess_output
665 def stream_callback(line: str, is_stderr: bool) -> None:
666 if stream_output:
667 if is_stderr:
668 print(f" {line}", file=sys.stderr)
669 else:
670 self.logger.info(f" {line}")
672 def on_start(process: subprocess.Popen[str]) -> None:
673 if issue_id:
674 with self._process_lock:
675 self._active_processes[issue_id] = process
677 def on_end(process: subprocess.Popen[str]) -> None:
678 if issue_id:
679 with self._process_lock:
680 self._active_processes.pop(issue_id, None)
682 return _run_claude_base(
683 command=command,
684 timeout=self.parallel_config.timeout_per_issue,
685 working_dir=working_dir,
686 stream_callback=stream_callback if stream_output else None,
687 on_process_start=on_start if issue_id else None,
688 on_process_end=on_end if issue_id else None,
689 idle_timeout=self.parallel_config.idle_timeout_per_issue,
690 on_usage=on_usage,
691 resume_session=resume_session,
692 )
694 def _run_with_continuation(
695 self,
696 command: str,
697 working_dir: Path,
698 issue_id: str | None = None,
699 max_continuations: int = 3,
700 context_limit: int = 200_000,
701 sentinel_threshold: float = 0.60,
702 guillotine_threshold: float = 0.90,
703 ) -> subprocess.CompletedProcess[str]:
704 """Run a Claude command with automatic continuation on context handoff.
706 Mirrors the E+G+J logic in issue_manager.run_with_continuation.
708 Args:
709 command: The command to run
710 working_dir: Directory (worktree) to run the command in
711 issue_id: Optional issue ID for subprocess tracking
712 max_continuations: Maximum number of continuation attempts
713 context_limit: Context window size in tokens
714 sentinel_threshold: Write sentinel when usage >= this fraction
715 guillotine_threshold: Trigger J-path when usage >= this fraction
717 Returns:
718 Combined CompletedProcess with all session outputs
719 """
720 all_stdout: list[str] = []
721 all_stderr: list[str] = []
722 current_command = command
723 continuation_count = 0
724 result: subprocess.CompletedProcess[str] = subprocess.CompletedProcess(
725 args=[], returncode=1, stdout="", stderr=""
726 )
727 tag = f"[{issue_id}]" if issue_id else "[worker]"
729 # Track token usage per-round for sentinel/guillotine thresholds
730 _last_input: list[int] = [0]
731 _last_output: list[int] = [0]
733 def _usage_tracker(input_tokens: int, output_tokens: int) -> None:
734 _last_input[0] = input_tokens
735 _last_output[0] = output_tokens
737 while continuation_count <= max_continuations:
738 result = self._run_claude_command(
739 current_command,
740 working_dir,
741 issue_id=issue_id,
742 on_usage=_usage_tracker,
743 )
745 all_stdout.append(result.stdout)
746 all_stderr.append(result.stderr)
748 # Standard path: Claude emitted CONTEXT_HANDOFF
749 if detect_context_handoff(result.stdout):
750 self.logger.info(f"{tag} Detected CONTEXT_HANDOFF signal")
752 prompt_content = read_continuation_prompt(working_dir)
753 if not prompt_content:
754 self.logger.warning(
755 f"{tag} Context handoff signaled but no continuation prompt found"
756 )
757 all_stderr.append("Handoff detected but no continuation prompt found")
758 result = subprocess.CompletedProcess(
759 args=result.args, returncode=1, stdout=result.stdout, stderr=result.stderr
760 )
761 break
763 if continuation_count >= max_continuations:
764 self.logger.warning(
765 f"{tag} Reached max continuations ({max_continuations}), stopping"
766 )
767 break
769 continuation_count += 1
770 self.logger.info(f"{tag} Starting continuation session #{continuation_count}")
771 current_command = f"{command} --resume"
772 continue
774 total_tokens = _last_input[0] + _last_output[0]
775 usage_ratio = total_tokens / context_limit if context_limit > 0 else 0.0
776 prompt_too_long = "prompt is too long" in (result.stderr or "").lower()
778 # Option J: guillotine — fresh session with transcript-summary prompt
779 if (
780 prompt_too_long or usage_ratio >= guillotine_threshold
781 ) and continuation_count < max_continuations:
782 trigger_reason = (
783 "Prompt is too long" if prompt_too_long else f"usage {usage_ratio:.0%}"
784 )
785 self.logger.warning(
786 f"{tag} Option J triggered ({trigger_reason}): spawning fresh session"
787 )
788 try:
789 guillotine_cmd = assemble_guillotine_prompt(
790 original_command=command,
791 captured_stdout="\n---CONTINUATION---\n".join(all_stdout),
792 token_stats={
793 "input_tokens": _last_input[0],
794 "output_tokens": _last_output[0],
795 "context_limit": context_limit,
796 "trigger_reason": trigger_reason,
797 },
798 )
799 except Exception as exc:
800 self.logger.warning(
801 f"{tag} Failed to assemble guillotine prompt ({exc}), using bare restart"
802 )
803 guillotine_cmd = command
804 continuation_count += 1
805 current_command = guillotine_cmd
806 _last_input[0] = 0
807 _last_output[0] = 0
808 continue
810 # Option E: read sentinel from a PREVIOUS session (must run before G writes
811 # the current-session sentinel to avoid immediately consuming our own write).
812 sentinel_data = read_sentinel(working_dir)
813 if sentinel_data is not None and continuation_count < max_continuations:
814 usage_pct = sentinel_data.get("usage_percent", int(usage_ratio * 100))
815 self.logger.info(
816 f"{tag} Sentinel detected ({usage_pct}% context used): "
817 "sending explicit handoff instruction"
818 )
819 continuation_count += 1
820 explicit_handoff_instruction = (
821 f"Context limit is approaching ({usage_pct}% of the context window is used). "
822 "Please run /ll:handoff RIGHT NOW to save your progress to "
823 ".ll/ll-continue-prompt.md, then output "
824 '"CONTEXT_HANDOFF: Ready for fresh session" to signal continuation.'
825 )
826 _last_input[0] = 0
827 _last_output[0] = 0
828 result = self._run_claude_command(
829 explicit_handoff_instruction,
830 working_dir,
831 issue_id=issue_id,
832 on_usage=_usage_tracker,
833 resume_session=True,
834 )
835 all_stdout.append(result.stdout)
836 all_stderr.append(result.stderr)
838 if detect_context_handoff(result.stdout):
839 self.logger.info(
840 f"{tag} CONTEXT_HANDOFF detected after explicit handoff instruction"
841 )
842 prompt_content = read_continuation_prompt(working_dir)
843 if prompt_content and continuation_count < max_continuations:
844 continuation_count += 1
845 self.logger.info(
846 f"{tag} Starting continuation session #{continuation_count}"
847 )
848 current_command = f"{command} --resume"
849 _last_input[0] = 0
850 _last_output[0] = 0
851 continue
852 break
854 # Option G (Python layer): write sentinel for the NEXT session.
855 # Placed after E-path so we don't immediately consume our own write.
856 if total_tokens > 0 and usage_ratio >= sentinel_threshold:
857 self.logger.info(
858 f"{tag} Writing context-handoff sentinel ({usage_ratio:.0%} context used)"
859 )
860 write_sentinel(working_dir, token_count=total_tokens, context_limit=context_limit)
862 # No handoff signal, no prior-session sentinel, no overflow — done
863 break
865 return subprocess.CompletedProcess(
866 args=result.args,
867 returncode=result.returncode,
868 stdout="\n---CONTINUATION---\n".join(all_stdout),
869 stderr="\n---CONTINUATION---\n".join(all_stderr),
870 )
872 def _get_changed_files(self, worktree_path: Path) -> list[str]:
873 """Get list of files changed in the worktree.
875 Args:
876 worktree_path: Path to the worktree
878 Returns:
879 List of changed file paths relative to repo root
880 """
881 result = subprocess.run(
882 ["git", "diff", "--name-only", self.parallel_config.base_branch, "HEAD"],
883 cwd=worktree_path,
884 capture_output=True,
885 text=True,
886 timeout=30,
887 )
889 if result.returncode != 0:
890 return []
892 return [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
894 def _update_branch_base(self, worktree_path: Path, issue_id: str) -> tuple[bool, str]:
895 """Fetch origin/main and rebase worker branch onto it.
897 This ensures the worker branch is based on the latest main before
898 merge coordination, preventing conflicts when main advances during
899 sprint execution (BUG-180).
901 Args:
902 worktree_path: Path to the worker's worktree
903 issue_id: Issue ID for logging
905 Returns:
906 Tuple of (success, error_message)
907 """
908 # Fetch latest base branch from configured remote (fall back to local if fetch fails)
909 base = self.parallel_config.base_branch
910 remote = self.parallel_config.remote_name
911 fetch_result = subprocess.run(
912 ["git", "fetch", remote, base],
913 cwd=worktree_path,
914 capture_output=True,
915 text=True,
916 timeout=60,
917 )
919 rebase_target = f"{remote}/{base}" if fetch_result.returncode == 0 else base
921 # Rebase current branch onto base (remote or local fallback)
922 rebase_result = subprocess.run(
923 ["git", "rebase", rebase_target],
924 cwd=worktree_path,
925 capture_output=True,
926 text=True,
927 timeout=120,
928 )
930 if rebase_result.returncode != 0:
931 # Abort the failed rebase
932 subprocess.run(
933 ["git", "rebase", "--abort"],
934 cwd=worktree_path,
935 capture_output=True,
936 timeout=10,
937 )
938 return False, f"Failed to rebase onto {rebase_target}: {rebase_result.stderr}"
940 self.logger.info(f"[{issue_id}] Rebased branch onto {rebase_target}")
941 return True, ""
943 def _verify_work_was_done(
944 self, changed_files: list[str], issue_id: str, issue_filename: str = ""
945 ) -> tuple[bool, str]:
946 """Verify that actual implementation work was done.
948 Uses the shared verify_work_was_done() function to check that changed
949 files include meaningful work, not just issue files or other artifacts.
951 Args:
952 changed_files: List of files changed during processing
953 issue_id: The issue ID being processed (unused, kept for compatibility)
954 issue_filename: Full issue filename (unused, kept for compatibility)
956 Returns:
957 Tuple of (success, error_message)
958 """
959 if not changed_files:
960 return False, "No files were changed during implementation"
962 # Check if code changes are required
963 if not self.parallel_config.require_code_changes:
964 return True, ""
966 # Use shared verification function
967 if verify_work_was_done(self.logger, changed_files):
968 return True, ""
970 # Generate descriptive error with actual excluded files
971 excluded_files = [
972 f
973 for f in changed_files
974 if f and any(f.startswith(excl) for excl in EXCLUDED_DIRECTORIES)
975 ]
976 if excluded_files:
977 files_preview = ", ".join(excluded_files[:5])
978 if len(excluded_files) > 5:
979 files_preview += f" (+{len(excluded_files) - 5} more)"
980 return False, f"Only excluded files modified: {files_preview}"
981 return False, "Only excluded files modified (e.g., .issues/, thoughts/)"
983 def _has_other_issue_id(self, file_lower: str, current_issue_id_lower: str) -> bool:
984 """Check if file contains a different issue ID than the current worker's.
986 This prevents cross-worker contamination where worker A detects worker B's
987 leaked file. When multiple workers run in parallel, their leaked files may
988 both appear in the main repo. Each worker should only clean up its own leaks.
990 Args:
991 file_lower: Lowercase file path to check
992 current_issue_id_lower: Lowercase issue ID of the current worker
994 Returns:
995 True if the file contains a different issue ID (belongs to another worker),
996 False if the file contains the current issue ID or no recognizable issue ID
997 """
998 # Pattern matches common issue ID formats: BUG-123, ENH-456, FEAT-789, EPIC-001
999 # Use non-capturing group (?:...) so findall returns full match, not group
1000 matches = re.findall(r"(?:bug|enh|feat|epic)-\d+", file_lower)
1002 if not matches:
1003 # No issue ID found - file doesn't belong to any specific worker
1004 return False
1006 # Check if any of the found issue IDs match the current worker
1007 for match in matches:
1008 if match == current_issue_id_lower:
1009 return False # File belongs to current worker
1011 # File has issue ID(s) but none match current worker - belongs to another worker
1012 return True
1014 def _detect_main_repo_leaks(self, issue_id: str, baseline_status: set[str]) -> list[str]:
1015 """Detect files incorrectly written to main repo instead of worktree.
1017 Claude Code may write files to the main repository instead of the
1018 worktree due to project root detection issues (see GitHub #8771).
1019 This method detects such leaks by comparing main repo status before
1020 and after worker execution.
1022 Args:
1023 issue_id: ID of the issue being processed (for pattern matching)
1024 baseline_status: Set of file paths from git status before worker started
1026 Returns:
1027 List of file paths that were leaked to main repo
1028 """
1029 # Get current status of main repo
1030 result = self._git_lock.run(
1031 ["status", "--porcelain"],
1032 cwd=self.repo_path,
1033 timeout=30,
1034 )
1036 if result.returncode != 0:
1037 return []
1039 current_files: set[str] = set()
1040 for line in result.stdout.strip().split("\n"):
1041 if not line or len(line) < 3:
1042 continue
1043 # Extract file path (after status codes and space)
1044 file_path = line[3:].strip()
1045 # Handle renamed files (old -> new)
1046 if " -> " in file_path:
1047 file_path = file_path.split(" -> ")[-1]
1048 current_files.add(file_path)
1050 # Find new files that appeared during worker execution
1051 new_files = current_files - baseline_status
1053 # Filter to files likely related to this issue
1054 issue_id_lower = issue_id.lower()
1055 leaked_files: list[str] = []
1057 # Build source prefix list: start with common fallbacks, then add configured dirs
1058 source_prefixes = ["backend/", "src/", "lib/", "tests/"]
1059 for dir_path in [self.br_config.project.src_dir, self.br_config.project.test_dir]:
1060 if dir_path:
1061 normalized = dir_path.rstrip("/") + "/"
1062 if normalized not in source_prefixes:
1063 source_prefixes.append(normalized)
1065 for file_path in new_files:
1066 # Skip state file (managed by orchestrator)
1067 if file_path.endswith(".parallel-manage-state.json"):
1068 continue
1069 # Skip .gitignore (may be modified by ll-parallel)
1070 if file_path == ".gitignore":
1071 continue
1073 # Check if file is related to this issue
1074 file_lower = file_path.lower()
1075 if issue_id_lower in file_lower:
1076 leaked_files.append(file_path)
1077 # Also catch source files that shouldn't be modified in main
1078 elif file_path.startswith(tuple(source_prefixes)):
1079 leaked_files.append(file_path)
1080 # Catch thoughts/plans files
1081 elif file_path.startswith("thoughts/"):
1082 leaked_files.append(file_path)
1083 # Catch issue files in any issue directory variant
1084 # Handles both .issues/ (with dot) and issues/ (without dot)
1085 # Only include files without a different issue ID - files WITH other issue IDs
1086 # belong to other workers running in parallel (cross-worker contamination)
1087 elif file_path.startswith((".issues/", "issues/")):
1088 if not self._has_other_issue_id(file_lower, issue_id_lower):
1089 leaked_files.append(file_path)
1091 return leaked_files
1093 def _cleanup_leaked_files(self, leaked_files: list[str]) -> int:
1094 """Discard leaked files from main repo working directory.
1096 Claude Code sometimes writes files to the main repo instead of the
1097 worktree. These files cause stash conflicts during merge operations.
1098 Since the actual work is preserved in the worktree branch, we can
1099 safely discard these leaked changes from the main repo.
1101 Args:
1102 leaked_files: List of file paths leaked to main repo
1104 Returns:
1105 Number of files successfully cleaned up
1106 """
1107 if not leaked_files:
1108 return 0
1110 cleaned = 0
1112 # Get status to determine which files are tracked vs untracked
1113 status_result = self._git_lock.run(
1114 ["status", "--porcelain", "--"] + leaked_files,
1115 cwd=self.repo_path,
1116 timeout=30,
1117 )
1119 tracked_files: list[str] = []
1120 untracked_files: list[str] = []
1122 for line in status_result.stdout.splitlines():
1123 if not line or len(line) < 3:
1124 continue
1125 status_code = line[:2]
1126 file_path = line[3:].split(" -> ")[-1].strip()
1128 if status_code.startswith("?"):
1129 # Untracked file - need to delete
1130 untracked_files.append(file_path)
1131 else:
1132 # Tracked file - can use git checkout to discard
1133 tracked_files.append(file_path)
1135 # Discard changes to tracked files
1136 if tracked_files:
1137 checkout_result = self._git_lock.run(
1138 ["checkout", "--"] + tracked_files,
1139 cwd=self.repo_path,
1140 timeout=30,
1141 )
1142 if checkout_result.returncode == 0:
1143 cleaned += len(tracked_files)
1144 else:
1145 self.logger.warning(
1146 f"Failed to discard tracked leaked files: {checkout_result.stderr}"
1147 )
1149 # Delete untracked files
1150 for file_path in untracked_files:
1151 full_path = self.repo_path / file_path
1152 try:
1153 if full_path.exists():
1154 full_path.unlink()
1155 cleaned += 1
1156 except OSError as e:
1157 self.logger.warning(f"Failed to delete leaked file {file_path}: {e}")
1159 # Fallback: directly delete files not reported by git status
1160 # This handles gitignored files that git status --porcelain doesn't show
1161 accounted_files = set(tracked_files + untracked_files)
1162 for file_path in leaked_files:
1163 if file_path not in accounted_files:
1164 full_path = self.repo_path / file_path
1165 if full_path.exists():
1166 try:
1167 full_path.unlink()
1168 cleaned += 1
1169 self.logger.info(f"Deleted gitignored leaked file: {file_path}")
1170 except OSError as e:
1171 self.logger.warning(
1172 f"Failed to delete gitignored leaked file {file_path}: {e}"
1173 )
1174 else:
1175 self.logger.debug(f"Leaked file not found (may have been moved): {file_path}")
1177 if cleaned > 0:
1178 self.logger.info(f"Cleaned up {cleaned} leaked file(s) from main repo")
1180 return cleaned
1182 def _get_main_repo_baseline(self) -> set[str]:
1183 """Get baseline of modified/untracked files in main repo.
1185 Returns:
1186 Set of file paths currently showing in git status
1187 """
1188 result = self._git_lock.run(
1189 ["status", "--porcelain"],
1190 cwd=self.repo_path,
1191 timeout=30,
1192 )
1194 if result.returncode != 0:
1195 return set()
1197 files: set[str] = set()
1198 for line in result.stdout.strip().split("\n"):
1199 if not line or len(line) < 3:
1200 continue
1201 file_path = line[3:].strip()
1202 if " -> " in file_path:
1203 file_path = file_path.split(" -> ")[-1]
1204 files.add(file_path)
1206 return files
1208 def _get_main_head_sha(self) -> str:
1209 """Get the current HEAD SHA of the main repo.
1211 Returns:
1212 HEAD SHA string, or empty string if unavailable
1213 """
1214 result = self._git_lock.run(
1215 ["rev-parse", "HEAD"],
1216 cwd=self.repo_path,
1217 timeout=10,
1218 )
1219 if result.returncode == 0:
1220 return result.stdout.strip()
1221 return ""
1223 def _detect_committed_leaks(self, baseline_head_sha: str) -> list[str]:
1224 """Detect commits made directly to main repo during worker execution.
1226 When Claude commits to the main repo instead of the worktree branch,
1227 the commits appear on main's history but the worktree has no changes.
1228 This method detects such leaked commits by comparing main's HEAD SHA
1229 before and after worker execution.
1231 Args:
1232 baseline_head_sha: HEAD SHA captured before worker started
1234 Returns:
1235 List of commit SHAs committed to main during worker execution,
1236 newest first. Empty list if no committed leaks detected.
1237 """
1238 if not baseline_head_sha:
1239 return []
1241 current_sha = self._get_main_head_sha()
1242 if not current_sha or current_sha == baseline_head_sha:
1243 return []
1245 # Get list of new commits on main since baseline
1246 result = self._git_lock.run(
1247 ["log", "--format=%H", f"{baseline_head_sha}..HEAD"],
1248 cwd=self.repo_path,
1249 timeout=30,
1250 )
1251 if result.returncode != 0:
1252 return []
1254 commits = [sha.strip() for sha in result.stdout.strip().split("\n") if sha.strip()]
1255 return commits
1257 def _recover_committed_leaks(
1258 self,
1259 leaked_commits: list[str],
1260 worktree_path: Path,
1261 baseline_head_sha: str,
1262 issue_id: str,
1263 ) -> bool:
1264 """Attempt to recover committed leaks by cherry-picking to worktree.
1266 When Claude commits directly to main instead of the worktree branch,
1267 we attempt to:
1268 1. Cherry-pick the leaked commits onto the worktree branch
1269 2. Reset main back to the baseline SHA (if safe to do so)
1271 This preserves the implementation work in the worktree while
1272 cleaning up the incorrect commits on main.
1274 Args:
1275 leaked_commits: Commit SHAs that leaked to main (newest first)
1276 worktree_path: Path to the worker's worktree
1277 baseline_head_sha: Main HEAD SHA before worker started
1278 issue_id: Issue ID for logging
1280 Returns:
1281 True if cherry-pick succeeded (main reset is attempted but
1282 not required for a True return value)
1283 """
1284 self.logger.info(
1285 f"[{issue_id}] Attempting recovery: cherry-picking {len(leaked_commits)} "
1286 f"commit(s) to worktree"
1287 )
1289 # Cherry-pick in chronological order (oldest first = reverse of log output)
1290 for sha in reversed(leaked_commits):
1291 result = subprocess.run(
1292 ["git", "cherry-pick", sha],
1293 cwd=worktree_path,
1294 capture_output=True,
1295 text=True,
1296 timeout=60,
1297 )
1298 if result.returncode != 0:
1299 subprocess.run(
1300 ["git", "cherry-pick", "--abort"],
1301 cwd=worktree_path,
1302 capture_output=True,
1303 timeout=10,
1304 )
1305 self.logger.warning(
1306 f"[{issue_id}] Cherry-pick of {sha[:8]} failed: {result.stderr.strip()}"
1307 )
1308 return False
1310 # Attempt to reset main to baseline (only if main hasn't advanced further)
1311 current_main_sha = self._get_main_head_sha()
1312 most_recent_leaked = leaked_commits[0] # Newest first
1313 if current_main_sha == most_recent_leaked:
1314 reset_result = self._git_lock.run(
1315 ["reset", "--hard", baseline_head_sha],
1316 cwd=self.repo_path,
1317 timeout=30,
1318 )
1319 if reset_result.returncode == 0:
1320 self.logger.info(f"[{issue_id}] Reset main to baseline {baseline_head_sha[:8]}")
1321 else:
1322 self.logger.warning(
1323 f"[{issue_id}] Cherry-pick succeeded but failed to reset main: "
1324 f"{reset_result.stderr.strip()}"
1325 )
1326 else:
1327 # main has advanced past the leaked commits — attempt surgical rebase
1328 # to excise only the leaked commits while preserving subsequent work
1329 self.logger.info(
1330 f"[{issue_id}] Main has advanced beyond leaked commits "
1331 f"({current_main_sha[:8]} != {most_recent_leaked[:8]}) — "
1332 f"attempting surgical rebase to excise leaked commits"
1333 )
1334 rebase_result = self._git_lock.run(
1335 ["rebase", "--onto", baseline_head_sha, most_recent_leaked],
1336 cwd=self.repo_path,
1337 timeout=60,
1338 )
1339 if rebase_result.returncode == 0:
1340 self.logger.info(f"[{issue_id}] Surgically removed leaked commits via rebase")
1341 else:
1342 self._git_lock.run(
1343 ["rebase", "--abort"],
1344 cwd=self.repo_path,
1345 timeout=10,
1346 )
1347 self.logger.warning(
1348 f"[{issue_id}] Surgical rebase failed — manual cleanup required: "
1349 f"{rebase_result.stderr.strip()}"
1350 )
1352 self.logger.info(
1353 f"[{issue_id}] Recovered {len(leaked_commits)} commit(s): "
1354 f"cherry-picked to worktree branch"
1355 )
1356 return True
1358 @property
1359 def active_count(self) -> int:
1360 """Number of currently active workers.
1362 Includes both workers with running futures AND workers whose futures
1363 are done but callbacks haven't completed yet.
1364 """
1365 with self._process_lock:
1366 running_futures = sum(1 for f in self._active_workers.values() if not f.done())
1367 with self._callback_lock:
1368 pending_callback_count = len(self._pending_callbacks)
1369 return running_futures + pending_callback_count
1371 def set_worker_stage(self, issue_id: str, stage: WorkerStage) -> None:
1372 """Update the stage of a worker.
1374 Args:
1375 issue_id: Issue ID being processed
1376 stage: New stage value
1377 """
1378 with self._process_lock:
1379 self._worker_stages[issue_id] = stage
1381 def get_worker_stage(self, issue_id: str) -> WorkerStage | None:
1382 """Get the current stage of a worker.
1384 Args:
1385 issue_id: Issue ID being processed
1387 Returns:
1388 Current stage, or None if issue not being tracked
1389 """
1390 with self._process_lock:
1391 return self._worker_stages.get(issue_id)
1393 def get_active_stages(self) -> dict[str, WorkerStage]:
1394 """Get all active worker stages.
1396 Returns:
1397 Dictionary mapping issue_id to current stage for active workers
1398 """
1399 with self._process_lock:
1400 # Only return workers that are actually active
1401 active_ids = set(self._active_workers.keys())
1402 return {
1403 issue_id: stage
1404 for issue_id, stage in self._worker_stages.items()
1405 if issue_id in active_ids
1406 }
1408 def remove_worker_stage(self, issue_id: str) -> None:
1409 """Remove a worker from stage tracking.
1411 Args:
1412 issue_id: Issue ID to remove
1413 """
1414 with self._process_lock:
1415 self._worker_stages.pop(issue_id, None)
1417 def cleanup_all_worktrees(self) -> None:
1418 """Clean up all worker worktrees."""
1419 worktree_base = self.repo_path / self.parallel_config.worktree_base
1420 if not worktree_base.exists():
1421 return
1423 from little_loops.worktree_utils import _is_ll_worktree
1425 for worktree_dir in worktree_base.iterdir():
1426 if worktree_dir.is_dir() and _is_ll_worktree(worktree_dir.name):
1427 self._cleanup_worktree(worktree_dir)
1429 self.logger.info("Cleaned up all worker worktrees")