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