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