Coverage for little_loops / parallel / worker_pool.py: 10%
538 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:08 -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.host_runner import resolve_host
23from little_loops.output_parsing import parse_ready_issue_output
24from little_loops.parallel.git_lock import GitLock
25from little_loops.parallel.types import ParallelConfig, WorkerResult, WorkerStage
26from little_loops.subprocess_utils import (
27 assemble_guillotine_prompt,
28 detect_context_handoff,
29 read_continuation_prompt,
30 read_sentinel,
31 write_sentinel,
32)
33from little_loops.subprocess_utils import (
34 run_claude_command as _run_claude_base,
35)
36from little_loops.work_verification import EXCLUDED_DIRECTORIES, verify_work_was_done
38if TYPE_CHECKING:
39 from little_loops.config import BRConfig
40 from little_loops.issue_parser import IssueInfo
41 from little_loops.logger import Logger
42 from little_loops.parallel.types import SprintWorkerContext
45class WorkerPool:
46 """Thread pool for processing issues in isolated git worktrees.
48 Each worker:
49 1. Creates a dedicated git worktree and branch
50 2. Runs issue validation and implementation via Claude CLI
51 3. Commits changes locally
52 4. Returns results for merge coordination
54 Example:
55 >>> pool = WorkerPool(parallel_config, br_config, logger)
56 >>> pool.start()
57 >>> future = pool.submit(issue_info)
58 >>> result = future.result() # WorkerResult
59 >>> pool.shutdown()
60 """
62 def __init__(
63 self,
64 parallel_config: ParallelConfig,
65 br_config: BRConfig,
66 logger: Logger,
67 repo_path: Path | None = None,
68 git_lock: GitLock | None = None,
69 ) -> None:
70 """Initialize the worker pool.
72 Args:
73 parallel_config: Parallel processing configuration
74 br_config: Project configuration (for category actions)
75 logger: Logger for worker output
76 repo_path: Path to the git repository (default: current directory)
77 git_lock: Shared lock for git operations (created if not provided)
78 """
79 self.parallel_config = parallel_config
80 self.br_config = br_config
81 self.logger = logger
82 self.repo_path = repo_path or Path.cwd()
83 self._git_lock = git_lock or GitLock(logger)
84 self._executor: ThreadPoolExecutor | None = None
85 self._active_workers: dict[str, Future[WorkerResult]] = {}
86 # Track active subprocesses for forceful termination on shutdown
87 self._active_processes: dict[str, subprocess.Popen[str]] = {}
88 # Track active worktree paths to prevent cleanup while in use (BUG-142)
89 self._active_worktrees: set[Path] = set()
90 self._process_lock = threading.Lock()
91 # Track callbacks currently executing
92 self._pending_callbacks: set[str] = set()
93 self._callback_lock = threading.Lock()
94 # Shutdown tracking for interrupted worker detection (ENH-036)
95 self._shutdown_requested = False
96 self._terminated_during_shutdown: set[str] = set()
97 # Track worker processing stages for progress visibility (ENH-262)
98 self._worker_stages: dict[str, WorkerStage] = {}
100 def start(self) -> None:
101 """Start the worker pool."""
102 if self._executor is not None:
103 return
105 # Ensure worktree base directory exists
106 worktree_base = self.repo_path / self.parallel_config.worktree_base
107 worktree_base.mkdir(parents=True, exist_ok=True)
109 self._executor = ThreadPoolExecutor(
110 max_workers=self.parallel_config.max_workers,
111 thread_name_prefix="issue-worker",
112 )
113 self.logger.info(f"Worker pool started with {self.parallel_config.max_workers} workers")
115 def shutdown(self, wait: bool = True) -> None:
116 """Shutdown the worker pool.
118 Args:
119 wait: Whether to wait for pending tasks to complete
120 """
121 if self._executor is None:
122 return
124 self.logger.info("Shutting down worker pool...")
126 # First, terminate all active subprocesses to unblock worker threads
127 if not wait:
128 self.terminate_all_processes()
130 self._executor.shutdown(wait=wait)
131 self._executor = None
133 def set_shutdown_requested(self, value: bool = True) -> None:
134 """Set the shutdown flag.
136 Called by orchestrator during shutdown to enable tracking of
137 workers that are terminated due to shutdown vs. actual failures.
138 """
139 self._shutdown_requested = value
141 def terminate_all_processes(self) -> None:
142 """Forcefully terminate all active subprocesses.
144 Called when we need to abort workers immediately,
145 such as on timeout or shutdown.
146 """
147 with self._process_lock:
148 for issue_id, process in list(self._active_processes.items()):
149 if process.poll() is None: # Still running
150 self.logger.warning(
151 f"Terminating subprocess for {issue_id} (PID {process.pid})"
152 )
153 # Track issues terminated during shutdown for interrupted detection (ENH-036)
154 if self._shutdown_requested:
155 self._terminated_during_shutdown.add(issue_id)
156 try:
157 # Send SIGTERM first for graceful termination
158 process.terminate()
159 try:
160 process.wait(timeout=5)
161 except subprocess.TimeoutExpired:
162 # Force kill if SIGTERM didn't work
163 self.logger.warning(f"Force killing {issue_id} (PID {process.pid})")
164 process.kill()
165 process.wait(timeout=2)
166 except Exception as e:
167 self.logger.error(f"Failed to terminate {issue_id}: {e}")
168 self._active_processes.clear()
170 def submit(
171 self,
172 issue: IssueInfo,
173 on_complete: Callable[[WorkerResult], None] | None = None,
174 ) -> Future[WorkerResult]:
175 """Submit an issue for processing.
177 Args:
178 issue: Issue to process
179 on_complete: Optional callback when processing completes
181 Returns:
182 Future that will contain the WorkerResult
183 """
184 if self._executor is None:
185 raise RuntimeError("Worker pool not started")
187 future = self._executor.submit(self._process_issue, issue)
188 with self._process_lock:
189 self._active_workers[issue.issue_id] = future
191 if on_complete:
192 future.add_done_callback(
193 lambda f: self._handle_completion(f, on_complete, issue.issue_id)
194 )
196 return future
198 def _handle_completion(
199 self,
200 future: Future[WorkerResult],
201 callback: Callable[[WorkerResult], None],
202 issue_id: str,
203 ) -> None:
204 """Handle worker completion and invoke callback."""
205 with self._callback_lock:
206 self._pending_callbacks.add(issue_id)
207 try:
208 try:
209 result = future.result()
210 except Exception as e:
211 self.logger.error(f"Worker future failed for {issue_id}: {e}")
212 result = WorkerResult(
213 issue_id=issue_id,
214 success=False,
215 branch_name="",
216 worktree_path=Path(),
217 error=f"Worker future failed: {e}",
218 )
219 # Set final stage based on result (ENH-262)
220 if result.success:
221 self.set_worker_stage(issue_id, WorkerStage.COMPLETED)
222 elif result.interrupted:
223 self.set_worker_stage(issue_id, WorkerStage.INTERRUPTED)
224 else:
225 self.set_worker_stage(issue_id, WorkerStage.FAILED)
226 try:
227 callback(result)
228 except Exception as e:
229 self.logger.error(f"Worker completion callback failed for {issue_id}: {e}")
230 finally:
231 with self._callback_lock:
232 self._pending_callbacks.discard(issue_id)
234 def _process_issue(self, issue: IssueInfo) -> WorkerResult:
235 """Process a single issue in an isolated worktree.
237 Args:
238 issue: Issue to process
240 Returns:
241 WorkerResult with processing outcome
242 """
243 start_time = time.time()
244 timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
245 if self.parallel_config.use_feature_branches:
246 from little_loops.issue_parser import slugify
248 branch_name = f"feature/{issue.issue_id.lower()}-{slugify(issue.title)}"
249 else:
250 branch_name = f"parallel/{issue.issue_id.lower()}-{timestamp}"
251 worktree_path = (
252 self.repo_path
253 / self.parallel_config.worktree_base
254 / f"worker-{issue.issue_id.lower()}-{timestamp}"
255 )
257 # Set initial stage for progress tracking (ENH-262)
258 self.set_worker_stage(issue.issue_id, WorkerStage.SETUP)
260 # Capture baseline of main repo status before worker starts
261 # Used to detect files incorrectly written to main repo
262 baseline_status = self._get_main_repo_baseline()
263 # Capture main HEAD SHA before worker starts to detect committed leaks
264 baseline_head_sha = self._get_main_head_sha()
266 try:
267 # Step 1: Create worktree with new branch
268 self._setup_worktree(
269 worktree_path,
270 branch_name,
271 base_branch=self.parallel_config.base_branch
272 if self.parallel_config.use_feature_branches
273 else None,
274 )
276 # Register worktree as active to prevent cleanup while in use (BUG-142)
277 with self._process_lock:
278 self._active_worktrees.add(worktree_path)
280 # Update stage for progress tracking (ENH-262)
281 self.set_worker_stage(issue.issue_id, WorkerStage.VALIDATING)
283 # Step 2: Run ready-issue validation
284 ready_cmd = self.parallel_config.get_ready_command(issue.issue_id)
285 ready_result = self._run_claude_command(
286 ready_cmd,
287 worktree_path,
288 issue_id=issue.issue_id,
289 )
291 # Check if worker was terminated during shutdown (ENH-036)
292 if issue.issue_id in self._terminated_during_shutdown:
293 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
294 return WorkerResult(
295 issue_id=issue.issue_id,
296 success=False,
297 interrupted=True,
298 branch_name=branch_name,
299 worktree_path=worktree_path,
300 duration=time.time() - start_time,
301 error="Interrupted during shutdown",
302 stdout=ready_result.stdout,
303 stderr=ready_result.stderr,
304 )
306 if ready_result.returncode != 0:
307 err_detail = ready_result.stderr or (ready_result.stdout or "")[:500]
308 return WorkerResult(
309 issue_id=issue.issue_id,
310 success=False,
311 branch_name=branch_name,
312 worktree_path=worktree_path,
313 duration=time.time() - start_time,
314 error=f"ready-issue failed: {err_detail}",
315 stdout=ready_result.stdout,
316 stderr=ready_result.stderr,
317 )
319 # Step 3: Parse ready-issue output and check verdict
320 ready_parsed = parse_ready_issue_output(ready_result.stdout)
322 # Handle CLOSE verdict - issue should not be implemented
323 if ready_parsed.get("should_close"):
324 return WorkerResult(
325 issue_id=issue.issue_id,
326 success=True, # Closure is a valid outcome
327 branch_name=branch_name,
328 worktree_path=worktree_path,
329 duration=time.time() - start_time,
330 should_close=True,
331 close_reason=ready_parsed.get("close_reason"),
332 close_status=ready_parsed.get("close_status"),
333 stdout=ready_result.stdout,
334 stderr=ready_result.stderr,
335 )
337 # Handle BLOCKED verdict - issue has open dependencies
338 if ready_parsed.get("is_blocked"):
339 return WorkerResult(
340 issue_id=issue.issue_id,
341 success=False,
342 was_blocked=True,
343 branch_name=branch_name,
344 worktree_path=worktree_path,
345 duration=time.time() - start_time,
346 error="ready-issue verdict: BLOCKED - open dependency detected",
347 stdout=ready_result.stdout,
348 stderr=ready_result.stderr,
349 )
351 # Handle NOT_READY verdict
352 if not ready_parsed["is_ready"]:
353 concerns = ready_parsed.get("concerns", [])
354 if concerns:
355 concern_msg = "; ".join(concerns)
356 elif ready_parsed["verdict"] == "UNKNOWN":
357 # For UNKNOWN verdicts, show a snippet of output for debugging
358 raw_out = (ready_result.stdout or "")[:200].strip()
359 concern_msg = (
360 f"Could not parse verdict. Output: {raw_out}..."
361 if raw_out
362 else "No output from ready-issue"
363 )
364 else:
365 concern_msg = "Issue not ready"
366 return WorkerResult(
367 issue_id=issue.issue_id,
368 success=False,
369 branch_name=branch_name,
370 worktree_path=worktree_path,
371 duration=time.time() - start_time,
372 error=f"ready-issue verdict: {ready_parsed['verdict']} - {concern_msg}",
373 stdout=ready_result.stdout,
374 stderr=ready_result.stderr,
375 )
377 # Track if issue was corrected (corrections stay in worktree)
378 was_corrected = ready_parsed.get("was_corrected", False)
379 corrections = ready_parsed.get("corrections", [])
381 # Update stage for progress tracking (ENH-262)
382 self.set_worker_stage(issue.issue_id, WorkerStage.IMPLEMENTING)
384 # Decision gate: invoke decide-issue when the issue requires a decision
385 if issue.decision_needed is True:
386 decide_cmd = self.parallel_config.get_decide_command(issue.issue_id)
387 decide_result = self._run_claude_command(
388 decide_cmd, worktree_path, issue_id=issue.issue_id
389 )
390 if decide_result.returncode != 0:
391 self.logger.warning(
392 f"[{issue.issue_id}] decide-issue command failed, "
393 "continuing to implementation anyway..."
394 )
396 # Step 4: Get action from BRConfig
397 action = self.br_config.get_category_action(issue.issue_type)
399 # Step 5: Run manage-issue implementation (with continuation support)
400 manage_cmd = self.parallel_config.get_manage_command(
401 issue.issue_type, action, issue.issue_id
402 )
403 manage_result = self._run_with_continuation(
404 manage_cmd,
405 worktree_path,
406 issue_id=issue.issue_id,
407 )
409 # Update stage for progress tracking (ENH-262)
410 self.set_worker_stage(issue.issue_id, WorkerStage.VERIFYING)
412 # Check if worker was terminated during shutdown (ENH-036)
413 if issue.issue_id in self._terminated_during_shutdown:
414 self.set_worker_stage(issue.issue_id, WorkerStage.INTERRUPTED)
415 return WorkerResult(
416 issue_id=issue.issue_id,
417 success=False,
418 interrupted=True,
419 branch_name=branch_name,
420 worktree_path=worktree_path,
421 duration=time.time() - start_time,
422 error="Interrupted during shutdown",
423 stdout=manage_result.stdout,
424 stderr=manage_result.stderr,
425 )
427 # Step 6: Get list of changed files in worktree
428 changed_files = self._get_changed_files(worktree_path)
430 # Step 8: Detect files leaked to main repo instead of worktree (unstaged)
431 leaked_files = self._detect_main_repo_leaks(issue.issue_id, baseline_status)
432 if leaked_files:
433 self.logger.warning(
434 f"{issue.issue_id} leaked {len(leaked_files)} file(s) to main repo: "
435 f"{leaked_files}"
436 )
437 # Clean up leaked files to prevent stash conflicts during merge.
438 # The actual work is preserved in the worktree branch.
439 self._cleanup_leaked_files(leaked_files)
441 # Step 8b: Detect commits made directly to main instead of worktree branch.
442 # If Claude committed to main (not the worktree), worktree will have no diff,
443 # causing work verification to fail. Attempt to recover by cherry-picking
444 # the leaked commits to the worktree and resetting main. (BUG-580)
445 committed_leaks = self._detect_committed_leaks(baseline_head_sha)
446 if committed_leaks:
447 self.logger.warning(
448 f"{issue.issue_id} committed {len(committed_leaks)} commit(s) directly "
449 f"to main instead of worktree: {[sha[:8] for sha in committed_leaks]}"
450 )
451 if not changed_files:
452 recovered = self._recover_committed_leaks(
453 committed_leaks, worktree_path, baseline_head_sha, issue.issue_id
454 )
455 if recovered:
456 changed_files = self._get_changed_files(worktree_path)
458 # Step 7: Verify actual work was done (after potential committed-leak recovery)
459 # Pass full filename for better doc-only keyword matching
460 issue_filename = issue.path.stem if issue.path else ""
461 work_verified, verification_error = self._verify_work_was_done(
462 changed_files, issue.issue_id, issue_filename
463 )
465 if manage_result.returncode != 0:
466 err_detail = manage_result.stderr or (manage_result.stdout or "")[:500]
467 return WorkerResult(
468 issue_id=issue.issue_id,
469 success=False,
470 branch_name=branch_name,
471 worktree_path=worktree_path,
472 changed_files=changed_files,
473 leaked_files=leaked_files,
474 duration=time.time() - start_time,
475 error=f"manage-issue failed: {err_detail}",
476 stdout=manage_result.stdout,
477 stderr=manage_result.stderr,
478 )
480 if not work_verified:
481 return WorkerResult(
482 issue_id=issue.issue_id,
483 success=False,
484 branch_name=branch_name,
485 worktree_path=worktree_path,
486 changed_files=changed_files,
487 leaked_files=leaked_files,
488 duration=time.time() - start_time,
489 error=verification_error,
490 stdout=manage_result.stdout,
491 stderr=manage_result.stderr,
492 )
494 # Step 9: Update branch base before merge (BUG-180)
495 # Fetch origin/main and rebase to ensure branch is based on latest main
496 base_updated, base_error = self._update_branch_base(worktree_path, issue.issue_id)
498 # Update stage for progress tracking (ENH-262)
499 self.set_worker_stage(issue.issue_id, WorkerStage.MERGING)
501 if not base_updated:
502 return WorkerResult(
503 issue_id=issue.issue_id,
504 success=False,
505 branch_name=branch_name,
506 worktree_path=worktree_path,
507 changed_files=changed_files,
508 leaked_files=leaked_files,
509 duration=time.time() - start_time,
510 error=base_error,
511 stdout=manage_result.stdout,
512 stderr=manage_result.stderr,
513 )
515 return WorkerResult(
516 issue_id=issue.issue_id,
517 success=True,
518 branch_name=branch_name,
519 worktree_path=worktree_path,
520 changed_files=changed_files,
521 leaked_files=leaked_files,
522 duration=time.time() - start_time,
523 error=None,
524 stdout=manage_result.stdout,
525 stderr=manage_result.stderr,
526 was_corrected=was_corrected,
527 corrections=corrections,
528 )
530 except Exception as e:
531 return WorkerResult(
532 issue_id=issue.issue_id,
533 success=False,
534 branch_name=branch_name,
535 worktree_path=worktree_path,
536 duration=time.time() - start_time,
537 error=str(e),
538 )
539 finally:
540 # Unregister worktree as no longer active (BUG-142)
541 with self._process_lock:
542 self._active_worktrees.discard(worktree_path)
544 def _setup_worktree(
545 self, worktree_path: Path, branch_name: str, base_branch: str | None = None
546 ) -> None:
547 """Create a git worktree with a new branch.
549 Args:
550 worktree_path: Path for the new worktree
551 branch_name: Name of the new branch
552 base_branch: Optional commit-ish to fork the branch from; None forks from HEAD.
553 """
554 from little_loops.worktree_utils import setup_worktree
556 setup_worktree(
557 repo_path=self.repo_path,
558 worktree_path=worktree_path,
559 branch_name=branch_name,
560 copy_files=self.parallel_config.worktree_copy_files,
561 logger=self.logger,
562 git_lock=self._git_lock,
563 base_branch=base_branch,
564 )
566 # Verify model if --show-model flag is set (requires API call)
567 if self.parallel_config.show_model:
568 model = self._detect_worktree_model_via_api(worktree_path)
569 if model:
570 self.logger.info(f" Using model: {model}")
571 else:
572 self.logger.warning(" Could not detect Claude CLI model")
574 def _detect_worktree_model_via_api(self, worktree_path: Path) -> str | None:
575 """Detect the model Claude will use by making an API call.
577 Runs a minimal Claude command with JSON output and parses the modelUsage
578 field to verify settings.local.json is being respected.
580 Args:
581 worktree_path: Path to the worktree to test
583 Returns:
584 Model name (e.g., "claude-sonnet-4-20250514") or None if unable to detect
585 """
586 try:
587 invocation = resolve_host().build_blocking_json(prompt="reply with just 'ok'")
588 # No-perm-skip preserved: this is a detection probe, not a real run.
589 args = [a for a in invocation.args if a != "--dangerously-skip-permissions"]
591 # Set environment to keep Claude in the project working directory (BUG-007)
592 # This ensures the first Claude CLI invocation in the worktree has the same
593 # project root behavior as subsequent invocations via run_claude_command()
594 env = os.environ.copy()
595 env["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1"
596 env.update(invocation.env)
598 result = subprocess.run(
599 [invocation.binary, *args],
600 cwd=worktree_path,
601 capture_output=True,
602 text=True,
603 timeout=30,
604 env=env,
605 )
606 if result.returncode == 0 and result.stdout.strip():
607 data: dict[str, Any] = json.loads(result.stdout.strip())
608 model_usage: dict[str, Any] = data.get("modelUsage", {})
609 # Return the first (primary) model from modelUsage
610 if model_usage:
611 return cast(str, next(iter(model_usage.keys())))
612 except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError):
613 pass
614 return None
616 def _cleanup_worktree(self, worktree_path: Path) -> None:
617 """Remove a git worktree and its associated branch.
619 Args:
620 worktree_path: Path to the worktree to remove
621 """
622 if not worktree_path.exists():
623 return
625 # Skip cleanup if worktree is actively in use by a running worker (BUG-142)
626 with self._process_lock:
627 if worktree_path in self._active_worktrees:
628 self.logger.warning(
629 f"Skipping cleanup of {worktree_path.name}: worktree is in active use"
630 )
631 return
633 # Only delete branches with the parallel/ prefix (legacy behavior for ll-parallel)
634 branch_result = subprocess.run(
635 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
636 cwd=worktree_path,
637 capture_output=True,
638 text=True,
639 )
640 branch_name = branch_result.stdout.strip() if branch_result.returncode == 0 else None
641 delete_branch = branch_name is not None and branch_name.startswith("parallel/")
643 from little_loops.worktree_utils import cleanup_worktree
645 cleanup_worktree(
646 worktree_path=worktree_path,
647 repo_path=self.repo_path,
648 logger=self.logger,
649 git_lock=self._git_lock,
650 delete_branch=delete_branch,
651 )
653 def _run_claude_command(
654 self,
655 command: str,
656 working_dir: Path,
657 issue_id: str | None = None,
658 on_usage: Callable[[int, int], None] | None = None,
659 resume_session: bool = False,
660 ) -> subprocess.CompletedProcess[str]:
661 """Run a Claude CLI command with real-time output streaming.
663 Args:
664 command: The command to run (e.g., "/ll:ready-issue BUG-123")
665 working_dir: Directory to run the command in
666 issue_id: Optional issue ID for subprocess tracking
667 on_usage: Optional usage callback for token tracking
668 resume_session: If True, passes --continue to the Claude CLI
670 Returns:
671 CompletedProcess with stdout and stderr
672 """
673 stream_output = self.parallel_config.stream_subprocess_output
675 def stream_callback(line: str, is_stderr: bool) -> None:
676 if stream_output:
677 if is_stderr:
678 print(f" {line}", file=sys.stderr)
679 else:
680 self.logger.info(f" {line}")
682 def on_start(process: subprocess.Popen[str]) -> None:
683 if issue_id:
684 with self._process_lock:
685 self._active_processes[issue_id] = process
687 def on_end(process: subprocess.Popen[str]) -> None:
688 if issue_id:
689 with self._process_lock:
690 self._active_processes.pop(issue_id, None)
692 return _run_claude_base(
693 command=command,
694 timeout=self.parallel_config.timeout_per_issue,
695 working_dir=working_dir,
696 stream_callback=stream_callback if stream_output else None,
697 on_process_start=on_start if issue_id else None,
698 on_process_end=on_end if issue_id else None,
699 idle_timeout=self.parallel_config.idle_timeout_per_issue,
700 on_usage=on_usage,
701 resume_session=resume_session,
702 )
704 def _check_issue_already_done(self, issue_id: str | None, working_dir: Path) -> bool:
705 """Check if the issue file's status indicates work is already complete.
707 Pre-continuation guard (BUG-1759): when the inner Claude session hits its
708 context limit but the issue was already marked done, skip the handoff and
709 return success rather than triggering an unnecessary handoff cycle.
711 Args:
712 issue_id: Issue identifier (e.g., "BUG-1759"), or None.
713 working_dir: Working directory (worktree) to search for issue files.
715 Returns:
716 True if the issue's status is 'done' or 'cancelled'.
717 """
718 if issue_id is None:
719 return False
720 issues_dir = working_dir / ".issues"
721 if not issues_dir.exists():
722 return False
723 try:
724 from little_loops.frontmatter import parse_frontmatter
726 # Search all category directories for the issue file
727 for cat_dir in issues_dir.iterdir():
728 if not cat_dir.is_dir():
729 continue
730 for f in cat_dir.iterdir():
731 if not f.is_file() or not f.suffix == ".md":
732 continue
733 if f"-{issue_id}-" in f.name or f.name.endswith(f"-{issue_id}.md"):
734 fm = parse_frontmatter(f.read_text(encoding="utf-8"))
735 return fm.get("status") in ("done", "cancelled")
736 return False
737 except Exception:
738 return False
740 def _run_with_continuation(
741 self,
742 command: str,
743 working_dir: Path,
744 issue_id: str | None = None,
745 max_continuations: int = 3,
746 context_limit: int = 200_000,
747 sentinel_threshold: float = 0.60,
748 guillotine_threshold: float = 0.90,
749 run_dir: str | None = None,
750 sprint_context: SprintWorkerContext | None = None,
751 ) -> subprocess.CompletedProcess[str]:
752 """Run a Claude command with automatic continuation on context handoff.
754 Mirrors the E+G+J logic in issue_manager.run_with_continuation.
756 Args:
757 command: The command to run
758 working_dir: Directory (worktree) to run the command in
759 issue_id: Optional issue ID for subprocess tracking
760 max_continuations: Maximum number of continuation attempts
761 context_limit: Context window size in tokens
762 sentinel_threshold: Write sentinel when usage >= this fraction
763 guillotine_threshold: Trigger J-path when usage >= this fraction
765 Returns:
766 Combined CompletedProcess with all session outputs
767 """
768 all_stdout: list[str] = []
769 all_stderr: list[str] = []
770 current_command = command
771 continuation_count = 0
772 result: subprocess.CompletedProcess[str] = subprocess.CompletedProcess(
773 args=[], returncode=1, stdout="", stderr=""
774 )
775 tag = f"[{issue_id}]" if issue_id else "[worker]"
777 # Track token usage per-round for sentinel/guillotine thresholds
778 _last_input: list[int] = [0]
779 _last_output: list[int] = [0]
781 def _usage_tracker(input_tokens: int, output_tokens: int) -> None:
782 _last_input[0] = input_tokens
783 _last_output[0] = output_tokens
785 while continuation_count <= max_continuations:
786 result = self._run_claude_command(
787 current_command,
788 working_dir,
789 issue_id=issue_id,
790 on_usage=_usage_tracker,
791 )
793 all_stdout.append(result.stdout)
794 all_stderr.append(result.stderr)
796 # Standard path: Claude emitted CONTEXT_HANDOFF
797 if detect_context_handoff(result.stdout):
798 self.logger.info(f"{tag} Detected CONTEXT_HANDOFF signal")
800 # Pre-continuation guard: if the issue is already done/cancelled,
801 # the work is complete — return success without signalling handoff
802 # so the outer FSM doesn't waste a handoff cycle on finished work.
803 already_done = self._check_issue_already_done(issue_id, working_dir)
804 if already_done:
805 self.logger.info(
806 f"{tag} Issue already done/cancelled; "
807 "skipping handoff and returning success"
808 )
809 result = subprocess.CompletedProcess(
810 args=result.args,
811 returncode=0,
812 stdout=result.stdout,
813 stderr=result.stderr,
814 )
815 break
817 # Forward CONTEXT_HANDOFF signal to stdout so the outer FSM's
818 # signal_detector can detect it via the existing HANDOFF_SIGNAL pattern.
819 handoff_message = "CONTEXT_HANDOFF: Ready for fresh session"
820 print(handoff_message)
821 self.logger.info(f"{tag} Forwarded handoff signal to stdout; exiting cleanly")
823 result = subprocess.CompletedProcess(
824 args=result.args,
825 returncode=0,
826 stdout=result.stdout + "\n" + handoff_message,
827 stderr=result.stderr,
828 )
829 break
831 total_tokens = _last_input[0] + _last_output[0]
832 usage_ratio = total_tokens / context_limit if context_limit > 0 else 0.0
833 prompt_too_long = "prompt is too long" in (result.stderr or "").lower()
835 # Option J: guillotine — fresh session.
836 # When run_dir is set (loop context), write a resume file and invoke /ll:resume.
837 # Otherwise fall back to the transcript-summary blob.
838 if (
839 prompt_too_long or usage_ratio >= guillotine_threshold
840 ) and continuation_count < max_continuations:
841 trigger_reason = (
842 "Prompt is too long" if prompt_too_long else f"usage {usage_ratio:.0%}"
843 )
844 self.logger.warning(
845 f"{tag} Option J triggered ({trigger_reason}): spawning fresh session"
846 )
847 if run_dir is not None:
848 try:
849 guillotine_file = Path(run_dir) / "guillotine-prompt.md"
850 guillotine_file.parent.mkdir(parents=True, exist_ok=True)
851 task_first_line = (command.strip().splitlines() or [""])[0]
852 sprint_framing = ""
853 if sprint_context is not None:
854 sprint_framing = (
855 f"## Sprint Worker Context\n"
856 f"You are a sprint worker. Process exactly ONE issue: "
857 f"{sprint_context.issue_id}\n"
858 f"After completing this issue, exit immediately — "
859 f"do NOT process other issues.\n"
860 f"Do NOT ask for further instructions. Exit with code 0.\n"
861 f"Branch: {sprint_context.branch}\n\n"
862 )
863 guillotine_file.write_text(
864 sprint_framing + f"## Intent\n"
865 f"Resume an interrupted automation session that hit the context limit.\n"
866 f"Original task: {task_first_line}\n"
867 f"Trigger reason: {trigger_reason} "
868 f"({_last_input[0] + _last_output[0]:,} / {context_limit:,} tokens)\n"
869 f"\n"
870 f"## Next Steps\n"
871 f"1. Check `git log` to see what was committed in the previous session\n"
872 f"2. Check the issue file status — if already done/cancelled, stop\n"
873 f"3. Review `.loops/tmp/scratch/` for partial progress notes\n"
874 f"4. Continue the original task from where it left off, "
875 f"skipping already-completed work\n",
876 encoding="utf-8",
877 )
878 guillotine_cmd = f"/ll:resume {guillotine_file}"
879 self.logger.info(f"{tag} Option J resume file written: {guillotine_file}")
880 except Exception as exc:
881 self.logger.warning(
882 f"{tag} Failed to write guillotine resume file ({exc}), "
883 "falling back to summary blob"
884 )
885 guillotine_cmd = command
886 else:
887 try:
888 guillotine_cmd = assemble_guillotine_prompt(
889 original_command=command,
890 captured_stdout="\n---CONTINUATION---\n".join(all_stdout),
891 token_stats={
892 "input_tokens": _last_input[0],
893 "output_tokens": _last_output[0],
894 "context_limit": context_limit,
895 "trigger_reason": trigger_reason,
896 },
897 sprint_context=sprint_context,
898 )
899 except Exception as exc:
900 self.logger.warning(
901 f"{tag} Failed to assemble guillotine prompt ({exc}), "
902 "using bare restart"
903 )
904 guillotine_cmd = command
905 continuation_count += 1
906 current_command = guillotine_cmd
907 _last_input[0] = 0
908 _last_output[0] = 0
909 continue
911 # Option E: read sentinel from a PREVIOUS session (must run before G writes
912 # the current-session sentinel to avoid immediately consuming our own write).
913 sentinel_data = read_sentinel(working_dir)
914 if sentinel_data is not None and continuation_count < max_continuations:
915 usage_pct = sentinel_data.get("usage_percent", int(usage_ratio * 100))
916 self.logger.info(
917 f"{tag} Sentinel detected ({usage_pct}% context used): "
918 "sending explicit handoff instruction"
919 )
920 continuation_count += 1
921 explicit_handoff_instruction = (
922 f"Context limit is approaching ({usage_pct}% of the context window is used). "
923 "Please run /ll:handoff RIGHT NOW to save your progress to "
924 ".ll/ll-continue-prompt.md, then output "
925 '"CONTEXT_HANDOFF: Ready for fresh session" to signal continuation.'
926 )
927 _last_input[0] = 0
928 _last_output[0] = 0
929 result = self._run_claude_command(
930 explicit_handoff_instruction,
931 working_dir,
932 issue_id=issue_id,
933 on_usage=_usage_tracker,
934 resume_session=True,
935 )
936 all_stdout.append(result.stdout)
937 all_stderr.append(result.stderr)
939 if detect_context_handoff(result.stdout):
940 self.logger.info(
941 f"{tag} CONTEXT_HANDOFF detected after explicit handoff instruction"
942 )
943 prompt_content = read_continuation_prompt(working_dir)
944 if prompt_content and continuation_count < max_continuations:
945 continuation_count += 1
946 self.logger.info(
947 f"{tag} Starting continuation session #{continuation_count}"
948 )
949 current_command = f"{command} --resume"
950 _last_input[0] = 0
951 _last_output[0] = 0
952 continue
953 break
955 # Option G (Python layer): write sentinel for the NEXT session.
956 # Placed after E-path so we don't immediately consume our own write.
957 if total_tokens > 0 and usage_ratio >= sentinel_threshold:
958 self.logger.info(
959 f"{tag} Writing context-handoff sentinel ({usage_ratio:.0%} context used)"
960 )
961 write_sentinel(working_dir, token_count=total_tokens, context_limit=context_limit)
963 # No handoff signal, no prior-session sentinel, no overflow — done
964 break
966 return subprocess.CompletedProcess(
967 args=result.args,
968 returncode=result.returncode,
969 stdout="\n---CONTINUATION---\n".join(all_stdout),
970 stderr="\n---CONTINUATION---\n".join(all_stderr),
971 )
973 def _get_changed_files(self, worktree_path: Path) -> list[str]:
974 """Get list of files changed in the worktree.
976 Args:
977 worktree_path: Path to the worktree
979 Returns:
980 List of changed file paths relative to repo root
981 """
982 result = subprocess.run(
983 ["git", "diff", "--name-only", self.parallel_config.base_branch, "HEAD"],
984 cwd=worktree_path,
985 capture_output=True,
986 text=True,
987 timeout=30,
988 )
990 if result.returncode != 0:
991 return []
993 return [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
995 def _update_branch_base(self, worktree_path: Path, issue_id: str) -> tuple[bool, str]:
996 """Fetch origin/main and rebase worker branch onto it.
998 This ensures the worker branch is based on the latest main before
999 merge coordination, preventing conflicts when main advances during
1000 sprint execution (BUG-180).
1002 Args:
1003 worktree_path: Path to the worker's worktree
1004 issue_id: Issue ID for logging
1006 Returns:
1007 Tuple of (success, error_message)
1008 """
1009 # Fetch latest base branch from configured remote (fall back to local if fetch fails)
1010 base = self.parallel_config.base_branch
1011 remote = self.parallel_config.remote_name
1012 fetch_result = subprocess.run(
1013 ["git", "fetch", remote, base],
1014 cwd=worktree_path,
1015 capture_output=True,
1016 text=True,
1017 timeout=60,
1018 )
1020 rebase_target = f"{remote}/{base}" if fetch_result.returncode == 0 else base
1022 # Rebase current branch onto base (remote or local fallback)
1023 rebase_result = subprocess.run(
1024 ["git", "rebase", rebase_target],
1025 cwd=worktree_path,
1026 capture_output=True,
1027 text=True,
1028 timeout=120,
1029 )
1031 if rebase_result.returncode != 0:
1032 # Abort the failed rebase
1033 subprocess.run(
1034 ["git", "rebase", "--abort"],
1035 cwd=worktree_path,
1036 capture_output=True,
1037 timeout=10,
1038 )
1039 return False, f"Failed to rebase onto {rebase_target}: {rebase_result.stderr}"
1041 self.logger.info(f"[{issue_id}] Rebased branch onto {rebase_target}")
1042 return True, ""
1044 def _verify_work_was_done(
1045 self, changed_files: list[str], issue_id: str, issue_filename: str = ""
1046 ) -> tuple[bool, str]:
1047 """Verify that actual implementation work was done.
1049 Uses the shared verify_work_was_done() function to check that changed
1050 files include meaningful work, not just issue files or other artifacts.
1052 Args:
1053 changed_files: List of files changed during processing
1054 issue_id: The issue ID being processed (unused, kept for compatibility)
1055 issue_filename: Full issue filename (unused, kept for compatibility)
1057 Returns:
1058 Tuple of (success, error_message)
1059 """
1060 if not changed_files:
1061 return False, "No files were changed during implementation"
1063 # Check if code changes are required
1064 if not self.parallel_config.require_code_changes:
1065 return True, ""
1067 # Use shared verification function
1068 if verify_work_was_done(self.logger, changed_files):
1069 return True, ""
1071 # Generate descriptive error with actual excluded files
1072 excluded_files = [
1073 f
1074 for f in changed_files
1075 if f and any(f.startswith(excl) for excl in EXCLUDED_DIRECTORIES)
1076 ]
1077 if excluded_files:
1078 files_preview = ", ".join(excluded_files[:5])
1079 if len(excluded_files) > 5:
1080 files_preview += f" (+{len(excluded_files) - 5} more)"
1081 return False, f"Only excluded files modified: {files_preview}"
1082 return False, "Only excluded files modified (e.g., .issues/, thoughts/)"
1084 def _has_other_issue_id(self, file_lower: str, current_issue_id_lower: str) -> bool:
1085 """Check if file contains a different issue ID than the current worker's.
1087 This prevents cross-worker contamination where worker A detects worker B's
1088 leaked file. When multiple workers run in parallel, their leaked files may
1089 both appear in the main repo. Each worker should only clean up its own leaks.
1091 Args:
1092 file_lower: Lowercase file path to check
1093 current_issue_id_lower: Lowercase issue ID of the current worker
1095 Returns:
1096 True if the file contains a different issue ID (belongs to another worker),
1097 False if the file contains the current issue ID or no recognizable issue ID
1098 """
1099 # Pattern matches common issue ID formats: BUG-123, ENH-456, FEAT-789, EPIC-001
1100 # Use non-capturing group (?:...) so findall returns full match, not group
1101 matches = re.findall(r"(?:bug|enh|feat|epic)-\d+", file_lower)
1103 if not matches:
1104 # No issue ID found - file doesn't belong to any specific worker
1105 return False
1107 # Check if any of the found issue IDs match the current worker
1108 for match in matches:
1109 if match == current_issue_id_lower:
1110 return False # File belongs to current worker
1112 # File has issue ID(s) but none match current worker - belongs to another worker
1113 return True
1115 def _detect_main_repo_leaks(self, issue_id: str, baseline_status: set[str]) -> list[str]:
1116 """Detect files incorrectly written to main repo instead of worktree.
1118 Claude Code may write files to the main repository instead of the
1119 worktree due to project root detection issues (see GitHub #8771).
1120 This method detects such leaks by comparing main repo status before
1121 and after worker execution.
1123 Args:
1124 issue_id: ID of the issue being processed (for pattern matching)
1125 baseline_status: Set of file paths from git status before worker started
1127 Returns:
1128 List of file paths that were leaked to main repo
1129 """
1130 # Get current status of main repo
1131 result = self._git_lock.run(
1132 ["status", "--porcelain"],
1133 cwd=self.repo_path,
1134 timeout=30,
1135 )
1137 if result.returncode != 0:
1138 return []
1140 current_files: set[str] = set()
1141 for line in result.stdout.strip().split("\n"):
1142 if not line or len(line) < 3:
1143 continue
1144 # Extract file path (after status codes and space)
1145 file_path = line[3:].strip()
1146 # Handle renamed files (old -> new)
1147 if " -> " in file_path:
1148 file_path = file_path.split(" -> ")[-1]
1149 current_files.add(file_path)
1151 # Find new files that appeared during worker execution
1152 new_files = current_files - baseline_status
1154 # Filter to files likely related to this issue
1155 issue_id_lower = issue_id.lower()
1156 leaked_files: list[str] = []
1158 # Build source prefix list: start with common fallbacks, then add configured dirs
1159 source_prefixes = ["backend/", "src/", "lib/", "tests/"]
1160 for dir_path in [self.br_config.project.src_dir, self.br_config.project.test_dir]:
1161 if dir_path:
1162 normalized = dir_path.rstrip("/") + "/"
1163 if normalized not in source_prefixes:
1164 source_prefixes.append(normalized)
1166 for file_path in new_files:
1167 # Skip state file (managed by orchestrator)
1168 if file_path.endswith(".parallel-manage-state.json"):
1169 continue
1170 # Skip .gitignore (may be modified by ll-parallel)
1171 if file_path == ".gitignore":
1172 continue
1174 # Check if file is related to this issue
1175 file_lower = file_path.lower()
1176 if issue_id_lower in file_lower:
1177 leaked_files.append(file_path)
1178 # Also catch source files that shouldn't be modified in main
1179 elif file_path.startswith(tuple(source_prefixes)):
1180 leaked_files.append(file_path)
1181 # Catch thoughts/plans files
1182 elif file_path.startswith("thoughts/"):
1183 leaked_files.append(file_path)
1184 # Catch issue files in any issue directory variant
1185 # Handles both .issues/ (with dot) and issues/ (without dot)
1186 # Only include files without a different issue ID - files WITH other issue IDs
1187 # belong to other workers running in parallel (cross-worker contamination)
1188 elif file_path.startswith((".issues/", "issues/")):
1189 if not self._has_other_issue_id(file_lower, issue_id_lower):
1190 leaked_files.append(file_path)
1192 return leaked_files
1194 def _cleanup_leaked_files(self, leaked_files: list[str]) -> int:
1195 """Discard leaked files from main repo working directory.
1197 Claude Code sometimes writes files to the main repo instead of the
1198 worktree. These files cause stash conflicts during merge operations.
1199 Since the actual work is preserved in the worktree branch, we can
1200 safely discard these leaked changes from the main repo.
1202 Args:
1203 leaked_files: List of file paths leaked to main repo
1205 Returns:
1206 Number of files successfully cleaned up
1207 """
1208 if not leaked_files:
1209 return 0
1211 cleaned = 0
1213 # Get status to determine which files are tracked vs untracked
1214 status_result = self._git_lock.run(
1215 ["status", "--porcelain", "--"] + leaked_files,
1216 cwd=self.repo_path,
1217 timeout=30,
1218 )
1220 tracked_files: list[str] = []
1221 untracked_files: list[str] = []
1223 for line in status_result.stdout.splitlines():
1224 if not line or len(line) < 3:
1225 continue
1226 status_code = line[:2]
1227 file_path = line[3:].split(" -> ")[-1].strip()
1229 if status_code.startswith("?"):
1230 # Untracked file - need to delete
1231 untracked_files.append(file_path)
1232 else:
1233 # Tracked file - can use git checkout to discard
1234 tracked_files.append(file_path)
1236 # Discard changes to tracked files
1237 if tracked_files:
1238 checkout_result = self._git_lock.run(
1239 ["checkout", "--"] + tracked_files,
1240 cwd=self.repo_path,
1241 timeout=30,
1242 )
1243 if checkout_result.returncode == 0:
1244 cleaned += len(tracked_files)
1245 else:
1246 self.logger.warning(
1247 f"Failed to discard tracked leaked files: {checkout_result.stderr}"
1248 )
1250 # Delete untracked files
1251 for file_path in untracked_files:
1252 full_path = self.repo_path / file_path
1253 try:
1254 if full_path.exists():
1255 full_path.unlink()
1256 cleaned += 1
1257 except OSError as e:
1258 self.logger.warning(f"Failed to delete leaked file {file_path}: {e}")
1260 # Fallback: directly delete files not reported by git status
1261 # This handles gitignored files that git status --porcelain doesn't show
1262 accounted_files = set(tracked_files + untracked_files)
1263 for file_path in leaked_files:
1264 if file_path not in accounted_files:
1265 full_path = self.repo_path / file_path
1266 if full_path.exists():
1267 try:
1268 full_path.unlink()
1269 cleaned += 1
1270 self.logger.info(f"Deleted gitignored leaked file: {file_path}")
1271 except OSError as e:
1272 self.logger.warning(
1273 f"Failed to delete gitignored leaked file {file_path}: {e}"
1274 )
1275 else:
1276 self.logger.debug(f"Leaked file not found (may have been moved): {file_path}")
1278 if cleaned > 0:
1279 self.logger.info(f"Cleaned up {cleaned} leaked file(s) from main repo")
1281 return cleaned
1283 def _get_main_repo_baseline(self) -> set[str]:
1284 """Get baseline of modified/untracked files in main repo.
1286 Returns:
1287 Set of file paths currently showing in git status
1288 """
1289 result = self._git_lock.run(
1290 ["status", "--porcelain"],
1291 cwd=self.repo_path,
1292 timeout=30,
1293 )
1295 if result.returncode != 0:
1296 return set()
1298 files: set[str] = set()
1299 for line in result.stdout.strip().split("\n"):
1300 if not line or len(line) < 3:
1301 continue
1302 file_path = line[3:].strip()
1303 if " -> " in file_path:
1304 file_path = file_path.split(" -> ")[-1]
1305 files.add(file_path)
1307 return files
1309 def _get_main_head_sha(self) -> str:
1310 """Get the current HEAD SHA of the main repo.
1312 Returns:
1313 HEAD SHA string, or empty string if unavailable
1314 """
1315 result = self._git_lock.run(
1316 ["rev-parse", "HEAD"],
1317 cwd=self.repo_path,
1318 timeout=10,
1319 )
1320 if result.returncode == 0:
1321 return result.stdout.strip()
1322 return ""
1324 def _detect_committed_leaks(self, baseline_head_sha: str) -> list[str]:
1325 """Detect commits made directly to main repo during worker execution.
1327 When Claude commits to the main repo instead of the worktree branch,
1328 the commits appear on main's history but the worktree has no changes.
1329 This method detects such leaked commits by comparing main's HEAD SHA
1330 before and after worker execution.
1332 Args:
1333 baseline_head_sha: HEAD SHA captured before worker started
1335 Returns:
1336 List of commit SHAs committed to main during worker execution,
1337 newest first. Empty list if no committed leaks detected.
1338 """
1339 if not baseline_head_sha:
1340 return []
1342 current_sha = self._get_main_head_sha()
1343 if not current_sha or current_sha == baseline_head_sha:
1344 return []
1346 # Get list of new commits on main since baseline
1347 result = self._git_lock.run(
1348 ["log", "--format=%H", f"{baseline_head_sha}..HEAD"],
1349 cwd=self.repo_path,
1350 timeout=30,
1351 )
1352 if result.returncode != 0:
1353 return []
1355 commits = [sha.strip() for sha in result.stdout.strip().split("\n") if sha.strip()]
1356 return commits
1358 def _recover_committed_leaks(
1359 self,
1360 leaked_commits: list[str],
1361 worktree_path: Path,
1362 baseline_head_sha: str,
1363 issue_id: str,
1364 ) -> bool:
1365 """Attempt to recover committed leaks by cherry-picking to worktree.
1367 When Claude commits directly to main instead of the worktree branch,
1368 we attempt to:
1369 1. Cherry-pick the leaked commits onto the worktree branch
1370 2. Reset main back to the baseline SHA (if safe to do so)
1372 This preserves the implementation work in the worktree while
1373 cleaning up the incorrect commits on main.
1375 Args:
1376 leaked_commits: Commit SHAs that leaked to main (newest first)
1377 worktree_path: Path to the worker's worktree
1378 baseline_head_sha: Main HEAD SHA before worker started
1379 issue_id: Issue ID for logging
1381 Returns:
1382 True if cherry-pick succeeded (main reset is attempted but
1383 not required for a True return value)
1384 """
1385 self.logger.info(
1386 f"[{issue_id}] Attempting recovery: cherry-picking {len(leaked_commits)} "
1387 f"commit(s) to worktree"
1388 )
1390 # Cherry-pick in chronological order (oldest first = reverse of log output)
1391 for sha in reversed(leaked_commits):
1392 result = subprocess.run(
1393 ["git", "cherry-pick", sha],
1394 cwd=worktree_path,
1395 capture_output=True,
1396 text=True,
1397 timeout=60,
1398 )
1399 if result.returncode != 0:
1400 subprocess.run(
1401 ["git", "cherry-pick", "--abort"],
1402 cwd=worktree_path,
1403 capture_output=True,
1404 timeout=10,
1405 )
1406 self.logger.warning(
1407 f"[{issue_id}] Cherry-pick of {sha[:8]} failed: {result.stderr.strip()}"
1408 )
1409 return False
1411 # Attempt to reset main to baseline (only if main hasn't advanced further)
1412 current_main_sha = self._get_main_head_sha()
1413 most_recent_leaked = leaked_commits[0] # Newest first
1414 if current_main_sha == most_recent_leaked:
1415 reset_result = self._git_lock.run(
1416 ["reset", "--hard", baseline_head_sha],
1417 cwd=self.repo_path,
1418 timeout=30,
1419 )
1420 if reset_result.returncode == 0:
1421 self.logger.info(f"[{issue_id}] Reset main to baseline {baseline_head_sha[:8]}")
1422 else:
1423 self.logger.warning(
1424 f"[{issue_id}] Cherry-pick succeeded but failed to reset main: "
1425 f"{reset_result.stderr.strip()}"
1426 )
1427 else:
1428 # main has advanced past the leaked commits — attempt surgical rebase
1429 # to excise only the leaked commits while preserving subsequent work
1430 self.logger.info(
1431 f"[{issue_id}] Main has advanced beyond leaked commits "
1432 f"({current_main_sha[:8]} != {most_recent_leaked[:8]}) — "
1433 f"attempting surgical rebase to excise leaked commits"
1434 )
1435 rebase_result = self._git_lock.run(
1436 ["rebase", "--onto", baseline_head_sha, most_recent_leaked],
1437 cwd=self.repo_path,
1438 timeout=60,
1439 )
1440 if rebase_result.returncode == 0:
1441 self.logger.info(f"[{issue_id}] Surgically removed leaked commits via rebase")
1442 else:
1443 self._git_lock.run(
1444 ["rebase", "--abort"],
1445 cwd=self.repo_path,
1446 timeout=10,
1447 )
1448 self.logger.warning(
1449 f"[{issue_id}] Surgical rebase failed — manual cleanup required: "
1450 f"{rebase_result.stderr.strip()}"
1451 )
1453 self.logger.info(
1454 f"[{issue_id}] Recovered {len(leaked_commits)} commit(s): "
1455 f"cherry-picked to worktree branch"
1456 )
1457 return True
1459 @property
1460 def active_count(self) -> int:
1461 """Number of currently active workers.
1463 Includes both workers with running futures AND workers whose futures
1464 are done but callbacks haven't completed yet.
1465 """
1466 with self._process_lock:
1467 running_futures = sum(1 for f in self._active_workers.values() if not f.done())
1468 with self._callback_lock:
1469 pending_callback_count = len(self._pending_callbacks)
1470 return running_futures + pending_callback_count
1472 def set_worker_stage(self, issue_id: str, stage: WorkerStage) -> None:
1473 """Update the stage of a worker.
1475 Args:
1476 issue_id: Issue ID being processed
1477 stage: New stage value
1478 """
1479 with self._process_lock:
1480 self._worker_stages[issue_id] = stage
1482 def get_worker_stage(self, issue_id: str) -> WorkerStage | None:
1483 """Get the current stage of a worker.
1485 Args:
1486 issue_id: Issue ID being processed
1488 Returns:
1489 Current stage, or None if issue not being tracked
1490 """
1491 with self._process_lock:
1492 return self._worker_stages.get(issue_id)
1494 def get_active_stages(self) -> dict[str, WorkerStage]:
1495 """Get all active worker stages.
1497 Returns:
1498 Dictionary mapping issue_id to current stage for active workers
1499 """
1500 with self._process_lock:
1501 # Only return workers that are actually active
1502 active_ids = set(self._active_workers.keys())
1503 return {
1504 issue_id: stage
1505 for issue_id, stage in self._worker_stages.items()
1506 if issue_id in active_ids
1507 }
1509 def remove_worker_stage(self, issue_id: str) -> None:
1510 """Remove a worker from stage tracking.
1512 Args:
1513 issue_id: Issue ID to remove
1514 """
1515 with self._process_lock:
1516 self._worker_stages.pop(issue_id, None)
1518 def cleanup_all_worktrees(self) -> None:
1519 """Clean up all worker worktrees."""
1520 worktree_base = self.repo_path / self.parallel_config.worktree_base
1521 if not worktree_base.exists():
1522 return
1524 from little_loops.worktree_utils import _is_ll_worktree
1526 for worktree_dir in worktree_base.iterdir():
1527 if worktree_dir.is_dir() and _is_ll_worktree(worktree_dir.name):
1528 self._cleanup_worktree(worktree_dir)
1530 self.logger.info("Cleaned up all worker worktrees")