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