Coverage for little_loops / cli / parallel.py: 11%
106 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""ll-parallel: Process issues concurrently using isolated git worktrees."""
3from __future__ import annotations
5import argparse
6import os
7import subprocess
8import sys
9from pathlib import Path
11from little_loops.cli.output import configure_output, use_color_enabled
12from little_loops.cli_args import (
13 add_context_limit_arg,
14 add_dry_run_arg,
15 add_handoff_threshold_arg,
16 add_idle_timeout_arg,
17 add_label_arg,
18 add_max_issues_arg,
19 add_only_arg,
20 add_quiet_arg,
21 add_resume_arg,
22 add_skip_arg,
23 add_skip_learning_gate_arg,
24 add_timeout_arg,
25 add_type_arg,
26 parse_issue_ids,
27 parse_issue_types,
28 parse_labels,
29 parse_priorities,
30)
31from little_loops.config import BRConfig
32from little_loops.logger import Logger
33from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context
36def main_parallel() -> int:
37 """Entry point for ll-parallel command.
39 Process issues concurrently using isolated git worktrees.
41 Returns:
42 Exit code (0 = success)
43 """
44 with cli_event_context(DEFAULT_DB_PATH, "ll-parallel", sys.argv[1:]):
45 parser = argparse.ArgumentParser(
46 description="Process issues concurrently using isolated git worktrees",
47 formatter_class=argparse.RawDescriptionHelpFormatter,
48 epilog="""
49Examples:
50 %(prog)s # Process with default workers
51 %(prog)s --workers 3 # Use 3 parallel workers
52 %(prog)s --dry-run # Preview what would be processed
53 %(prog)s --priority P1,P2 # Only process P1 and P2 issues
54 %(prog)s --cleanup # Clean up worktrees and exit
55 %(prog)s --cleanup-orphans # Clean orphaned worktrees (liveness-aware, skips live-process worktrees)
56 %(prog)s --cleanup-orphans --dry-run # Preview orphan cleanup
57 %(prog)s --prune-merged-branches # Delete merged feature/* branches
58 %(prog)s --prune-merged-branches --dry-run # Preview what would be pruned
59 %(prog)s --stream-output # Stream Claude CLI output in real-time
60 %(prog)s --only BUG-001,BUG-002 # Process only specific issues
61 %(prog)s --skip BUG-003 # Skip specific issues
62 %(prog)s --type BUG # Process only bugs
63 %(prog)s --type BUG,ENH # Process bugs and enhancements
64""",
65 )
67 # Parallel-specific arguments (--workers, not --max-workers)
68 parser.add_argument(
69 "--workers",
70 "-w",
71 type=int,
72 default=None,
73 help="Number of parallel workers (default: from config or 2)",
74 )
75 parser.add_argument(
76 "--priority",
77 "-p",
78 type=str,
79 default=None,
80 help="Comma-separated priorities to process (default: all)",
81 )
82 parser.add_argument(
83 "--worktree-base",
84 type=Path,
85 default=None,
86 help="Base directory for git worktrees",
87 )
88 parser.add_argument(
89 "--cleanup",
90 "-c",
91 action="store_true",
92 help="Clean up all worktrees and exit",
93 )
94 parser.add_argument(
95 "--cleanup-orphans",
96 action="store_true",
97 help=(
98 "Remove orphaned worktrees from interrupted runs (liveness-aware: skips "
99 "worktrees owned by live processes). Use with --dry-run to preview."
100 ),
101 )
102 parser.add_argument(
103 "--prune-merged-branches",
104 action="store_true",
105 help=(
106 "Delete local feature/* branches already merged into the base branch. "
107 "Use --dry-run to preview. "
108 "Squash/rebase-merged branches require gh CLI for detection."
109 ),
110 )
111 parser.add_argument(
112 "--merge-pending",
113 action="store_true",
114 help="Attempt to merge pending work from previous interrupted runs",
115 )
116 parser.add_argument(
117 "--clean-start",
118 action="store_true",
119 help="Remove all worktrees and start fresh (skip pending work check)",
120 )
121 parser.add_argument(
122 "--ignore-pending",
123 action="store_true",
124 help="Report pending work but continue without merging",
125 )
126 parser.add_argument(
127 "--stream-output",
128 action="store_true",
129 help="Stream Claude CLI subprocess output to console",
130 )
131 parser.add_argument(
132 "--show-model",
133 action="store_true",
134 help="Make API call to verify and display model on worktree setup",
135 )
136 parser.add_argument(
137 "--feature-branches",
138 action=argparse.BooleanOptionalAction,
139 default=None,
140 help="Enable/disable feature-branch mode for this run (overrides config)",
141 )
142 parser.add_argument(
143 "--overlap-detection",
144 action="store_true",
145 help="Enable pre-flight overlap detection to reduce merge conflicts (ENH-143)",
146 )
147 parser.add_argument(
148 "--warn-only",
149 action="store_true",
150 help="With --overlap-detection, warn about overlaps instead of serializing",
151 )
153 parser.add_argument(
154 "--verbose",
155 "-v",
156 action="store_true",
157 help="Enable verbose output (default when --quiet is not set)",
158 )
160 # Add common arguments from shared module
161 add_dry_run_arg(parser)
162 add_resume_arg(parser)
163 add_timeout_arg(parser)
164 add_idle_timeout_arg(parser)
165 add_handoff_threshold_arg(parser)
166 add_context_limit_arg(parser)
167 add_quiet_arg(parser)
168 add_only_arg(parser)
169 add_skip_arg(parser)
170 add_type_arg(parser)
171 add_label_arg(parser)
172 add_skip_learning_gate_arg(parser)
174 # Add max-issues and config individually (different help text needed)
175 add_max_issues_arg(parser)
176 parser.add_argument(
177 "--config",
178 "-C",
179 type=Path,
180 default=None,
181 help="Path to project root",
182 )
184 args = parser.parse_args()
186 project_root = args.config or Path.cwd()
187 config = BRConfig(project_root)
188 configure_output(config.cli)
190 logger = Logger(verbose=args.verbose or not args.quiet, use_color=use_color_enabled())
192 # Handle cleanup mode
193 if args.cleanup:
194 from little_loops.parallel import WorkerPool
196 parallel_config = config.create_parallel_config()
197 pool = WorkerPool(parallel_config, config, logger, project_root)
198 pool.cleanup_all_worktrees()
199 logger.success("Cleanup complete")
200 return 0
202 # Handle cleanup-orphans mode (liveness-aware orphan cleanup)
203 if args.cleanup_orphans:
204 from little_loops.parallel import ParallelOrchestrator
206 parallel_config = config.create_parallel_config(dry_run=args.dry_run)
207 orchestrator = ParallelOrchestrator(
208 parallel_config=parallel_config,
209 br_config=config,
210 repo_path=project_root,
211 verbose=args.verbose or not args.quiet,
212 )
213 if args.dry_run:
214 logger.info("[DRY RUN] No changes will be made.")
215 orchestrator._cleanup_orphaned_worktrees(dry_run=args.dry_run)
216 logger.success("Orphan cleanup complete")
217 return 0
219 # Build priority filter (validates against VALID_PRIORITIES)
220 priority_filter = parse_priorities(args.priority)
222 if args.handoff_threshold is not None:
223 if not (1 <= args.handoff_threshold <= 100):
224 parser.error("--handoff-threshold must be between 1 and 100")
225 os.environ["LL_HANDOFF_THRESHOLD"] = str(args.handoff_threshold)
227 if args.context_limit is not None:
228 if args.context_limit < 50000:
229 parser.error("--context-limit must be at least 50000")
230 os.environ["LL_CONTEXT_LIMIT"] = str(args.context_limit)
232 # Parse issue ID filters
233 only_ids = parse_issue_ids(args.only)
234 skip_ids = parse_issue_ids(args.skip)
235 type_prefixes = parse_issue_types(args.type)
236 label_filter = parse_labels(args.label)
238 # Detect current branch for rebase/merge operations (BUG-439)
239 _branch_result = subprocess.run(
240 ["git", "rev-parse", "--abbrev-ref", "HEAD"],
241 capture_output=True,
242 text=True,
243 cwd=project_root,
244 )
245 _base_branch = _branch_result.stdout.strip() if _branch_result.returncode == 0 else "main"
247 # Create parallel config with CLI overrides
248 parallel_config = config.create_parallel_config(
249 max_workers=args.workers,
250 priority_filter=sorted(priority_filter) if priority_filter is not None else None,
251 label_filter=label_filter,
252 max_issues=args.max_issues,
253 dry_run=args.dry_run,
254 timeout_seconds=args.timeout,
255 idle_timeout_per_issue=args.idle_timeout,
256 stream_output=args.stream_output if args.stream_output else None,
257 show_model=args.show_model if args.show_model else None,
258 only_ids=only_ids,
259 skip_ids=skip_ids,
260 type_prefixes=type_prefixes,
261 merge_pending=args.merge_pending,
262 clean_start=args.clean_start,
263 ignore_pending=args.ignore_pending,
264 overlap_detection=args.overlap_detection,
265 serialize_overlapping=not args.warn_only,
266 base_branch=_base_branch,
267 use_feature_branches=args.feature_branches,
268 skip_learning_gate=args.skip_learning_gate,
269 )
271 # Handle prune mode
272 if args.prune_merged_branches:
273 from little_loops.parallel import WorkerPool
275 pool = WorkerPool(parallel_config, config, logger, project_root)
276 if args.dry_run:
277 logger.info("[DRY RUN] No branches will be deleted.")
278 pruned, skipped = pool.prune_merged_feature_branches(
279 base_branch=parallel_config.base_branch,
280 dry_run=args.dry_run,
281 )
282 if pruned:
283 verb = "would delete" if args.dry_run else "deleted"
284 logger.success(
285 f"Pruned {len(pruned)} merged feature branch(es) ({verb}): {', '.join(pruned)}"
286 )
287 else:
288 logger.info("No merged feature branches found to prune.")
289 if skipped:
290 logger.warning(f"Could not delete {len(skipped)} branch(es): {', '.join(skipped)}")
291 return 0
293 # Delete state file if not resuming
294 if not args.resume:
295 state_file = config.get_parallel_state_file()
296 if state_file.exists():
297 state_file.unlink()
299 # Create and run orchestrator
300 from little_loops.events import EventBus
301 from little_loops.parallel import ParallelOrchestrator
303 event_bus = EventBus()
304 from little_loops.extension import wire_extensions
305 from little_loops.transport import wire_transports
307 wire_extensions(event_bus, config.extensions)
308 wire_transports(event_bus, config.events)
309 orchestrator = ParallelOrchestrator(
310 parallel_config=parallel_config,
311 br_config=config,
312 repo_path=project_root,
313 verbose=args.verbose or not args.quiet,
314 event_bus=event_bus,
315 )
317 return orchestrator.run()