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