Coverage for little_loops / parallel / worker_pool.py: 10%
524 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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 ) -> subprocess.CompletedProcess[str]:
739 """Run a Claude command with automatic continuation on context handoff.
741 Mirrors the E+G+J logic in issue_manager.run_with_continuation.
743 Args:
744 command: The command to run
745 working_dir: Directory (worktree) to run the command in
746 issue_id: Optional issue ID for subprocess tracking
747 max_continuations: Maximum number of continuation attempts
748 context_limit: Context window size in tokens
749 sentinel_threshold: Write sentinel when usage >= this fraction
750 guillotine_threshold: Trigger J-path when usage >= this fraction
752 Returns:
753 Combined CompletedProcess with all session outputs
754 """
755 all_stdout: list[str] = []
756 all_stderr: list[str] = []
757 current_command = command
758 continuation_count = 0
759 result: subprocess.CompletedProcess[str] = subprocess.CompletedProcess(
760 args=[], returncode=1, stdout="", stderr=""
761 )
762 tag = f"[{issue_id}]" if issue_id else "[worker]"
764 # Track token usage per-round for sentinel/guillotine thresholds
765 _last_input: list[int] = [0]
766 _last_output: list[int] = [0]
768 def _usage_tracker(input_tokens: int, output_tokens: int) -> None:
769 _last_input[0] = input_tokens
770 _last_output[0] = output_tokens
772 while continuation_count <= max_continuations:
773 result = self._run_claude_command(
774 current_command,
775 working_dir,
776 issue_id=issue_id,
777 on_usage=_usage_tracker,
778 )
780 all_stdout.append(result.stdout)
781 all_stderr.append(result.stderr)
783 # Standard path: Claude emitted CONTEXT_HANDOFF
784 if detect_context_handoff(result.stdout):
785 self.logger.info(f"{tag} Detected CONTEXT_HANDOFF signal")
787 # Pre-continuation guard: if the issue is already done/cancelled,
788 # the work is complete — return success without signalling handoff
789 # so the outer FSM doesn't waste a handoff cycle on finished work.
790 already_done = self._check_issue_already_done(issue_id, working_dir)
791 if already_done:
792 self.logger.info(
793 f"{tag} Issue already done/cancelled; "
794 "skipping handoff and returning success"
795 )
796 result = subprocess.CompletedProcess(
797 args=result.args,
798 returncode=0,
799 stdout=result.stdout,
800 stderr=result.stderr,
801 )
802 break
804 # Forward CONTEXT_HANDOFF signal to stdout so the outer FSM's
805 # signal_detector can detect it via the existing HANDOFF_SIGNAL pattern.
806 handoff_message = "CONTEXT_HANDOFF: Ready for fresh session"
807 print(handoff_message)
808 self.logger.info(f"{tag} Forwarded handoff signal to stdout; exiting cleanly")
810 result = subprocess.CompletedProcess(
811 args=result.args,
812 returncode=0,
813 stdout=result.stdout + "\n" + handoff_message,
814 stderr=result.stderr,
815 )
816 break
818 total_tokens = _last_input[0] + _last_output[0]
819 usage_ratio = total_tokens / context_limit if context_limit > 0 else 0.0
820 prompt_too_long = "prompt is too long" in (result.stderr or "").lower()
822 # Option J: guillotine — fresh session with transcript-summary prompt
823 if (
824 prompt_too_long or usage_ratio >= guillotine_threshold
825 ) and continuation_count < max_continuations:
826 trigger_reason = (
827 "Prompt is too long" if prompt_too_long else f"usage {usage_ratio:.0%}"
828 )
829 self.logger.warning(
830 f"{tag} Option J triggered ({trigger_reason}): spawning fresh session"
831 )
832 try:
833 guillotine_cmd = assemble_guillotine_prompt(
834 original_command=command,
835 captured_stdout="\n---CONTINUATION---\n".join(all_stdout),
836 token_stats={
837 "input_tokens": _last_input[0],
838 "output_tokens": _last_output[0],
839 "context_limit": context_limit,
840 "trigger_reason": trigger_reason,
841 },
842 )
843 except Exception as exc:
844 self.logger.warning(
845 f"{tag} Failed to assemble guillotine prompt ({exc}), using bare restart"
846 )
847 guillotine_cmd = command
848 continuation_count += 1
849 current_command = guillotine_cmd
850 _last_input[0] = 0
851 _last_output[0] = 0
852 continue
854 # Option E: read sentinel from a PREVIOUS session (must run before G writes
855 # the current-session sentinel to avoid immediately consuming our own write).
856 sentinel_data = read_sentinel(working_dir)
857 if sentinel_data is not None and continuation_count < max_continuations:
858 usage_pct = sentinel_data.get("usage_percent", int(usage_ratio * 100))
859 self.logger.info(
860 f"{tag} Sentinel detected ({usage_pct}% context used): "
861 "sending explicit handoff instruction"
862 )
863 continuation_count += 1
864 explicit_handoff_instruction = (
865 f"Context limit is approaching ({usage_pct}% of the context window is used). "
866 "Please run /ll:handoff RIGHT NOW to save your progress to "
867 ".ll/ll-continue-prompt.md, then output "
868 '"CONTEXT_HANDOFF: Ready for fresh session" to signal continuation.'
869 )
870 _last_input[0] = 0
871 _last_output[0] = 0
872 result = self._run_claude_command(
873 explicit_handoff_instruction,
874 working_dir,
875 issue_id=issue_id,
876 on_usage=_usage_tracker,
877 resume_session=True,
878 )
879 all_stdout.append(result.stdout)
880 all_stderr.append(result.stderr)
882 if detect_context_handoff(result.stdout):
883 self.logger.info(
884 f"{tag} CONTEXT_HANDOFF detected after explicit handoff instruction"
885 )
886 prompt_content = read_continuation_prompt(working_dir)
887 if prompt_content and continuation_count < max_continuations:
888 continuation_count += 1
889 self.logger.info(
890 f"{tag} Starting continuation session #{continuation_count}"
891 )
892 current_command = f"{command} --resume"
893 _last_input[0] = 0
894 _last_output[0] = 0
895 continue
896 break
898 # Option G (Python layer): write sentinel for the NEXT session.
899 # Placed after E-path so we don't immediately consume our own write.
900 if total_tokens > 0 and usage_ratio >= sentinel_threshold:
901 self.logger.info(
902 f"{tag} Writing context-handoff sentinel ({usage_ratio:.0%} context used)"
903 )
904 write_sentinel(working_dir, token_count=total_tokens, context_limit=context_limit)
906 # No handoff signal, no prior-session sentinel, no overflow — done
907 break
909 return subprocess.CompletedProcess(
910 args=result.args,
911 returncode=result.returncode,
912 stdout="\n---CONTINUATION---\n".join(all_stdout),
913 stderr="\n---CONTINUATION---\n".join(all_stderr),
914 )
916 def _get_changed_files(self, worktree_path: Path) -> list[str]:
917 """Get list of files changed in the worktree.
919 Args:
920 worktree_path: Path to the worktree
922 Returns:
923 List of changed file paths relative to repo root
924 """
925 result = subprocess.run(
926 ["git", "diff", "--name-only", self.parallel_config.base_branch, "HEAD"],
927 cwd=worktree_path,
928 capture_output=True,
929 text=True,
930 timeout=30,
931 )
933 if result.returncode != 0:
934 return []
936 return [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
938 def _update_branch_base(self, worktree_path: Path, issue_id: str) -> tuple[bool, str]:
939 """Fetch origin/main and rebase worker branch onto it.
941 This ensures the worker branch is based on the latest main before
942 merge coordination, preventing conflicts when main advances during
943 sprint execution (BUG-180).
945 Args:
946 worktree_path: Path to the worker's worktree
947 issue_id: Issue ID for logging
949 Returns:
950 Tuple of (success, error_message)
951 """
952 # Fetch latest base branch from configured remote (fall back to local if fetch fails)
953 base = self.parallel_config.base_branch
954 remote = self.parallel_config.remote_name
955 fetch_result = subprocess.run(
956 ["git", "fetch", remote, base],
957 cwd=worktree_path,
958 capture_output=True,
959 text=True,
960 timeout=60,
961 )
963 rebase_target = f"{remote}/{base}" if fetch_result.returncode == 0 else base
965 # Rebase current branch onto base (remote or local fallback)
966 rebase_result = subprocess.run(
967 ["git", "rebase", rebase_target],
968 cwd=worktree_path,
969 capture_output=True,
970 text=True,
971 timeout=120,
972 )
974 if rebase_result.returncode != 0:
975 # Abort the failed rebase
976 subprocess.run(
977 ["git", "rebase", "--abort"],
978 cwd=worktree_path,
979 capture_output=True,
980 timeout=10,
981 )
982 return False, f"Failed to rebase onto {rebase_target}: {rebase_result.stderr}"
984 self.logger.info(f"[{issue_id}] Rebased branch onto {rebase_target}")
985 return True, ""
987 def _verify_work_was_done(
988 self, changed_files: list[str], issue_id: str, issue_filename: str = ""
989 ) -> tuple[bool, str]:
990 """Verify that actual implementation work was done.
992 Uses the shared verify_work_was_done() function to check that changed
993 files include meaningful work, not just issue files or other artifacts.
995 Args:
996 changed_files: List of files changed during processing
997 issue_id: The issue ID being processed (unused, kept for compatibility)
998 issue_filename: Full issue filename (unused, kept for compatibility)
1000 Returns:
1001 Tuple of (success, error_message)
1002 """
1003 if not changed_files:
1004 return False, "No files were changed during implementation"
1006 # Check if code changes are required
1007 if not self.parallel_config.require_code_changes:
1008 return True, ""
1010 # Use shared verification function
1011 if verify_work_was_done(self.logger, changed_files):
1012 return True, ""
1014 # Generate descriptive error with actual excluded files
1015 excluded_files = [
1016 f
1017 for f in changed_files
1018 if f and any(f.startswith(excl) for excl in EXCLUDED_DIRECTORIES)
1019 ]
1020 if excluded_files:
1021 files_preview = ", ".join(excluded_files[:5])
1022 if len(excluded_files) > 5:
1023 files_preview += f" (+{len(excluded_files) - 5} more)"
1024 return False, f"Only excluded files modified: {files_preview}"
1025 return False, "Only excluded files modified (e.g., .issues/, thoughts/)"
1027 def _has_other_issue_id(self, file_lower: str, current_issue_id_lower: str) -> bool:
1028 """Check if file contains a different issue ID than the current worker's.
1030 This prevents cross-worker contamination where worker A detects worker B's
1031 leaked file. When multiple workers run in parallel, their leaked files may
1032 both appear in the main repo. Each worker should only clean up its own leaks.
1034 Args:
1035 file_lower: Lowercase file path to check
1036 current_issue_id_lower: Lowercase issue ID of the current worker
1038 Returns:
1039 True if the file contains a different issue ID (belongs to another worker),
1040 False if the file contains the current issue ID or no recognizable issue ID
1041 """
1042 # Pattern matches common issue ID formats: BUG-123, ENH-456, FEAT-789, EPIC-001
1043 # Use non-capturing group (?:...) so findall returns full match, not group
1044 matches = re.findall(r"(?:bug|enh|feat|epic)-\d+", file_lower)
1046 if not matches:
1047 # No issue ID found - file doesn't belong to any specific worker
1048 return False
1050 # Check if any of the found issue IDs match the current worker
1051 for match in matches:
1052 if match == current_issue_id_lower:
1053 return False # File belongs to current worker
1055 # File has issue ID(s) but none match current worker - belongs to another worker
1056 return True
1058 def _detect_main_repo_leaks(self, issue_id: str, baseline_status: set[str]) -> list[str]:
1059 """Detect files incorrectly written to main repo instead of worktree.
1061 Claude Code may write files to the main repository instead of the
1062 worktree due to project root detection issues (see GitHub #8771).
1063 This method detects such leaks by comparing main repo status before
1064 and after worker execution.
1066 Args:
1067 issue_id: ID of the issue being processed (for pattern matching)
1068 baseline_status: Set of file paths from git status before worker started
1070 Returns:
1071 List of file paths that were leaked to main repo
1072 """
1073 # Get current status of main repo
1074 result = self._git_lock.run(
1075 ["status", "--porcelain"],
1076 cwd=self.repo_path,
1077 timeout=30,
1078 )
1080 if result.returncode != 0:
1081 return []
1083 current_files: set[str] = set()
1084 for line in result.stdout.strip().split("\n"):
1085 if not line or len(line) < 3:
1086 continue
1087 # Extract file path (after status codes and space)
1088 file_path = line[3:].strip()
1089 # Handle renamed files (old -> new)
1090 if " -> " in file_path:
1091 file_path = file_path.split(" -> ")[-1]
1092 current_files.add(file_path)
1094 # Find new files that appeared during worker execution
1095 new_files = current_files - baseline_status
1097 # Filter to files likely related to this issue
1098 issue_id_lower = issue_id.lower()
1099 leaked_files: list[str] = []
1101 # Build source prefix list: start with common fallbacks, then add configured dirs
1102 source_prefixes = ["backend/", "src/", "lib/", "tests/"]
1103 for dir_path in [self.br_config.project.src_dir, self.br_config.project.test_dir]:
1104 if dir_path:
1105 normalized = dir_path.rstrip("/") + "/"
1106 if normalized not in source_prefixes:
1107 source_prefixes.append(normalized)
1109 for file_path in new_files:
1110 # Skip state file (managed by orchestrator)
1111 if file_path.endswith(".parallel-manage-state.json"):
1112 continue
1113 # Skip .gitignore (may be modified by ll-parallel)
1114 if file_path == ".gitignore":
1115 continue
1117 # Check if file is related to this issue
1118 file_lower = file_path.lower()
1119 if issue_id_lower in file_lower:
1120 leaked_files.append(file_path)
1121 # Also catch source files that shouldn't be modified in main
1122 elif file_path.startswith(tuple(source_prefixes)):
1123 leaked_files.append(file_path)
1124 # Catch thoughts/plans files
1125 elif file_path.startswith("thoughts/"):
1126 leaked_files.append(file_path)
1127 # Catch issue files in any issue directory variant
1128 # Handles both .issues/ (with dot) and issues/ (without dot)
1129 # Only include files without a different issue ID - files WITH other issue IDs
1130 # belong to other workers running in parallel (cross-worker contamination)
1131 elif file_path.startswith((".issues/", "issues/")):
1132 if not self._has_other_issue_id(file_lower, issue_id_lower):
1133 leaked_files.append(file_path)
1135 return leaked_files
1137 def _cleanup_leaked_files(self, leaked_files: list[str]) -> int:
1138 """Discard leaked files from main repo working directory.
1140 Claude Code sometimes writes files to the main repo instead of the
1141 worktree. These files cause stash conflicts during merge operations.
1142 Since the actual work is preserved in the worktree branch, we can
1143 safely discard these leaked changes from the main repo.
1145 Args:
1146 leaked_files: List of file paths leaked to main repo
1148 Returns:
1149 Number of files successfully cleaned up
1150 """
1151 if not leaked_files:
1152 return 0
1154 cleaned = 0
1156 # Get status to determine which files are tracked vs untracked
1157 status_result = self._git_lock.run(
1158 ["status", "--porcelain", "--"] + leaked_files,
1159 cwd=self.repo_path,
1160 timeout=30,
1161 )
1163 tracked_files: list[str] = []
1164 untracked_files: list[str] = []
1166 for line in status_result.stdout.splitlines():
1167 if not line or len(line) < 3:
1168 continue
1169 status_code = line[:2]
1170 file_path = line[3:].split(" -> ")[-1].strip()
1172 if status_code.startswith("?"):
1173 # Untracked file - need to delete
1174 untracked_files.append(file_path)
1175 else:
1176 # Tracked file - can use git checkout to discard
1177 tracked_files.append(file_path)
1179 # Discard changes to tracked files
1180 if tracked_files:
1181 checkout_result = self._git_lock.run(
1182 ["checkout", "--"] + tracked_files,
1183 cwd=self.repo_path,
1184 timeout=30,
1185 )
1186 if checkout_result.returncode == 0:
1187 cleaned += len(tracked_files)
1188 else:
1189 self.logger.warning(
1190 f"Failed to discard tracked leaked files: {checkout_result.stderr}"
1191 )
1193 # Delete untracked files
1194 for file_path in untracked_files:
1195 full_path = self.repo_path / file_path
1196 try:
1197 if full_path.exists():
1198 full_path.unlink()
1199 cleaned += 1
1200 except OSError as e:
1201 self.logger.warning(f"Failed to delete leaked file {file_path}: {e}")
1203 # Fallback: directly delete files not reported by git status
1204 # This handles gitignored files that git status --porcelain doesn't show
1205 accounted_files = set(tracked_files + untracked_files)
1206 for file_path in leaked_files:
1207 if file_path not in accounted_files:
1208 full_path = self.repo_path / file_path
1209 if full_path.exists():
1210 try:
1211 full_path.unlink()
1212 cleaned += 1
1213 self.logger.info(f"Deleted gitignored leaked file: {file_path}")
1214 except OSError as e:
1215 self.logger.warning(
1216 f"Failed to delete gitignored leaked file {file_path}: {e}"
1217 )
1218 else:
1219 self.logger.debug(f"Leaked file not found (may have been moved): {file_path}")
1221 if cleaned > 0:
1222 self.logger.info(f"Cleaned up {cleaned} leaked file(s) from main repo")
1224 return cleaned
1226 def _get_main_repo_baseline(self) -> set[str]:
1227 """Get baseline of modified/untracked files in main repo.
1229 Returns:
1230 Set of file paths currently showing in git status
1231 """
1232 result = self._git_lock.run(
1233 ["status", "--porcelain"],
1234 cwd=self.repo_path,
1235 timeout=30,
1236 )
1238 if result.returncode != 0:
1239 return set()
1241 files: set[str] = set()
1242 for line in result.stdout.strip().split("\n"):
1243 if not line or len(line) < 3:
1244 continue
1245 file_path = line[3:].strip()
1246 if " -> " in file_path:
1247 file_path = file_path.split(" -> ")[-1]
1248 files.add(file_path)
1250 return files
1252 def _get_main_head_sha(self) -> str:
1253 """Get the current HEAD SHA of the main repo.
1255 Returns:
1256 HEAD SHA string, or empty string if unavailable
1257 """
1258 result = self._git_lock.run(
1259 ["rev-parse", "HEAD"],
1260 cwd=self.repo_path,
1261 timeout=10,
1262 )
1263 if result.returncode == 0:
1264 return result.stdout.strip()
1265 return ""
1267 def _detect_committed_leaks(self, baseline_head_sha: str) -> list[str]:
1268 """Detect commits made directly to main repo during worker execution.
1270 When Claude commits to the main repo instead of the worktree branch,
1271 the commits appear on main's history but the worktree has no changes.
1272 This method detects such leaked commits by comparing main's HEAD SHA
1273 before and after worker execution.
1275 Args:
1276 baseline_head_sha: HEAD SHA captured before worker started
1278 Returns:
1279 List of commit SHAs committed to main during worker execution,
1280 newest first. Empty list if no committed leaks detected.
1281 """
1282 if not baseline_head_sha:
1283 return []
1285 current_sha = self._get_main_head_sha()
1286 if not current_sha or current_sha == baseline_head_sha:
1287 return []
1289 # Get list of new commits on main since baseline
1290 result = self._git_lock.run(
1291 ["log", "--format=%H", f"{baseline_head_sha}..HEAD"],
1292 cwd=self.repo_path,
1293 timeout=30,
1294 )
1295 if result.returncode != 0:
1296 return []
1298 commits = [sha.strip() for sha in result.stdout.strip().split("\n") if sha.strip()]
1299 return commits
1301 def _recover_committed_leaks(
1302 self,
1303 leaked_commits: list[str],
1304 worktree_path: Path,
1305 baseline_head_sha: str,
1306 issue_id: str,
1307 ) -> bool:
1308 """Attempt to recover committed leaks by cherry-picking to worktree.
1310 When Claude commits directly to main instead of the worktree branch,
1311 we attempt to:
1312 1. Cherry-pick the leaked commits onto the worktree branch
1313 2. Reset main back to the baseline SHA (if safe to do so)
1315 This preserves the implementation work in the worktree while
1316 cleaning up the incorrect commits on main.
1318 Args:
1319 leaked_commits: Commit SHAs that leaked to main (newest first)
1320 worktree_path: Path to the worker's worktree
1321 baseline_head_sha: Main HEAD SHA before worker started
1322 issue_id: Issue ID for logging
1324 Returns:
1325 True if cherry-pick succeeded (main reset is attempted but
1326 not required for a True return value)
1327 """
1328 self.logger.info(
1329 f"[{issue_id}] Attempting recovery: cherry-picking {len(leaked_commits)} "
1330 f"commit(s) to worktree"
1331 )
1333 # Cherry-pick in chronological order (oldest first = reverse of log output)
1334 for sha in reversed(leaked_commits):
1335 result = subprocess.run(
1336 ["git", "cherry-pick", sha],
1337 cwd=worktree_path,
1338 capture_output=True,
1339 text=True,
1340 timeout=60,
1341 )
1342 if result.returncode != 0:
1343 subprocess.run(
1344 ["git", "cherry-pick", "--abort"],
1345 cwd=worktree_path,
1346 capture_output=True,
1347 timeout=10,
1348 )
1349 self.logger.warning(
1350 f"[{issue_id}] Cherry-pick of {sha[:8]} failed: {result.stderr.strip()}"
1351 )
1352 return False
1354 # Attempt to reset main to baseline (only if main hasn't advanced further)
1355 current_main_sha = self._get_main_head_sha()
1356 most_recent_leaked = leaked_commits[0] # Newest first
1357 if current_main_sha == most_recent_leaked:
1358 reset_result = self._git_lock.run(
1359 ["reset", "--hard", baseline_head_sha],
1360 cwd=self.repo_path,
1361 timeout=30,
1362 )
1363 if reset_result.returncode == 0:
1364 self.logger.info(f"[{issue_id}] Reset main to baseline {baseline_head_sha[:8]}")
1365 else:
1366 self.logger.warning(
1367 f"[{issue_id}] Cherry-pick succeeded but failed to reset main: "
1368 f"{reset_result.stderr.strip()}"
1369 )
1370 else:
1371 # main has advanced past the leaked commits — attempt surgical rebase
1372 # to excise only the leaked commits while preserving subsequent work
1373 self.logger.info(
1374 f"[{issue_id}] Main has advanced beyond leaked commits "
1375 f"({current_main_sha[:8]} != {most_recent_leaked[:8]}) — "
1376 f"attempting surgical rebase to excise leaked commits"
1377 )
1378 rebase_result = self._git_lock.run(
1379 ["rebase", "--onto", baseline_head_sha, most_recent_leaked],
1380 cwd=self.repo_path,
1381 timeout=60,
1382 )
1383 if rebase_result.returncode == 0:
1384 self.logger.info(f"[{issue_id}] Surgically removed leaked commits via rebase")
1385 else:
1386 self._git_lock.run(
1387 ["rebase", "--abort"],
1388 cwd=self.repo_path,
1389 timeout=10,
1390 )
1391 self.logger.warning(
1392 f"[{issue_id}] Surgical rebase failed — manual cleanup required: "
1393 f"{rebase_result.stderr.strip()}"
1394 )
1396 self.logger.info(
1397 f"[{issue_id}] Recovered {len(leaked_commits)} commit(s): "
1398 f"cherry-picked to worktree branch"
1399 )
1400 return True
1402 @property
1403 def active_count(self) -> int:
1404 """Number of currently active workers.
1406 Includes both workers with running futures AND workers whose futures
1407 are done but callbacks haven't completed yet.
1408 """
1409 with self._process_lock:
1410 running_futures = sum(1 for f in self._active_workers.values() if not f.done())
1411 with self._callback_lock:
1412 pending_callback_count = len(self._pending_callbacks)
1413 return running_futures + pending_callback_count
1415 def set_worker_stage(self, issue_id: str, stage: WorkerStage) -> None:
1416 """Update the stage of a worker.
1418 Args:
1419 issue_id: Issue ID being processed
1420 stage: New stage value
1421 """
1422 with self._process_lock:
1423 self._worker_stages[issue_id] = stage
1425 def get_worker_stage(self, issue_id: str) -> WorkerStage | None:
1426 """Get the current stage of a worker.
1428 Args:
1429 issue_id: Issue ID being processed
1431 Returns:
1432 Current stage, or None if issue not being tracked
1433 """
1434 with self._process_lock:
1435 return self._worker_stages.get(issue_id)
1437 def get_active_stages(self) -> dict[str, WorkerStage]:
1438 """Get all active worker stages.
1440 Returns:
1441 Dictionary mapping issue_id to current stage for active workers
1442 """
1443 with self._process_lock:
1444 # Only return workers that are actually active
1445 active_ids = set(self._active_workers.keys())
1446 return {
1447 issue_id: stage
1448 for issue_id, stage in self._worker_stages.items()
1449 if issue_id in active_ids
1450 }
1452 def remove_worker_stage(self, issue_id: str) -> None:
1453 """Remove a worker from stage tracking.
1455 Args:
1456 issue_id: Issue ID to remove
1457 """
1458 with self._process_lock:
1459 self._worker_stages.pop(issue_id, None)
1461 def cleanup_all_worktrees(self) -> None:
1462 """Clean up all worker worktrees."""
1463 worktree_base = self.repo_path / self.parallel_config.worktree_base
1464 if not worktree_base.exists():
1465 return
1467 from little_loops.worktree_utils import _is_ll_worktree
1469 for worktree_dir in worktree_base.iterdir():
1470 if worktree_dir.is_dir() and _is_ll_worktree(worktree_dir.name):
1471 self._cleanup_worktree(worktree_dir)
1473 self.logger.info("Cleaned up all worker worktrees")