Coverage for little_loops / parallel / types.py: 85%
152 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:08 -0500
1"""Type definitions for parallel issue processing.
3Provides dataclasses for worker results, merge requests, orchestrator state,
4and parallel configuration. Reuses IssueInfo from issue_parser.py.
5"""
7from __future__ import annotations
9import time
10from dataclasses import dataclass, field
11from enum import Enum
12from pathlib import Path
13from typing import TYPE_CHECKING, Any
15if TYPE_CHECKING:
16 from little_loops.issue_parser import IssueInfo
19@dataclass
20class QueuedIssue:
21 """Issue in priority queue with ordering support.
23 Uses __lt__ for priority queue comparison. Lower priority number = higher priority.
24 Within same priority level, earlier timestamp wins (FIFO).
26 Attributes:
27 priority: Numeric priority (0=P0, 1=P1, etc.)
28 issue_info: The parsed issue information
29 timestamp: When the issue was queued (for FIFO ordering)
30 """
32 priority: int
33 issue_info: IssueInfo
34 timestamp: float = field(default_factory=time.time)
36 def __lt__(self, other: QueuedIssue) -> bool:
37 """Compare issues for priority queue ordering."""
38 if self.priority != other.priority:
39 return self.priority < other.priority
40 return self.timestamp < other.timestamp
42 def to_dict(self) -> dict[str, Any]:
43 """Convert to dictionary for JSON serialization."""
44 return {
45 "priority": self.priority,
46 "issue_info": self.issue_info.to_dict(),
47 "timestamp": self.timestamp,
48 }
51@dataclass
52class WorkerResult:
53 """Result from a worker processing an issue.
55 Attributes:
56 issue_id: ID of the processed issue
57 success: Whether processing succeeded
58 branch_name: Git branch created for this issue
59 worktree_path: Path to the worker's git worktree
60 changed_files: List of files modified during processing
61 leaked_files: Files incorrectly written to main repo instead of worktree
62 duration: Processing time in seconds
63 error: Error message if processing failed
64 stdout: Captured standard output
65 stderr: Captured standard error
66 was_corrected: Whether the issue file was auto-corrected
67 corrections: List of corrections made during validation (for pattern analysis)
68 should_close: Whether the issue should be closed (e.g., already fixed, invalid)
69 close_reason: Reason code for closure (e.g., "already_fixed")
70 close_status: Status text for closure (e.g., "Closed - Already Fixed")
71 interrupted: Whether the worker was interrupted during shutdown
72 """
74 issue_id: str
75 success: bool
76 branch_name: str
77 worktree_path: Path
78 changed_files: list[str] = field(default_factory=list)
79 leaked_files: list[str] = field(default_factory=list)
80 duration: float = 0.0
81 error: str | None = None
82 stdout: str = ""
83 stderr: str = ""
84 was_corrected: bool = False
85 corrections: list[str] = field(default_factory=list)
86 should_close: bool = False
87 close_reason: str | None = None
88 close_status: str | None = None
89 was_blocked: bool = False
90 interrupted: bool = False
92 def to_dict(self) -> dict[str, Any]:
93 """Convert to dictionary for JSON serialization."""
94 return {
95 "issue_id": self.issue_id,
96 "success": self.success,
97 "branch_name": self.branch_name,
98 "worktree_path": str(self.worktree_path),
99 "changed_files": self.changed_files,
100 "leaked_files": self.leaked_files,
101 "duration": self.duration,
102 "error": self.error,
103 "stdout": self.stdout,
104 "stderr": self.stderr,
105 "was_corrected": self.was_corrected,
106 "corrections": self.corrections,
107 "should_close": self.should_close,
108 "close_reason": self.close_reason,
109 "close_status": self.close_status,
110 "was_blocked": self.was_blocked,
111 "interrupted": self.interrupted,
112 }
114 @classmethod
115 def from_dict(cls, data: dict[str, Any]) -> WorkerResult:
116 """Create from dictionary (JSON deserialization)."""
117 return cls(
118 issue_id=data["issue_id"],
119 success=data["success"],
120 branch_name=data["branch_name"],
121 worktree_path=Path(data["worktree_path"]),
122 changed_files=data.get("changed_files", []),
123 leaked_files=data.get("leaked_files", []),
124 duration=data.get("duration", 0.0),
125 error=data.get("error"),
126 stdout=data.get("stdout", ""),
127 stderr=data.get("stderr", ""),
128 was_corrected=data.get("was_corrected", False),
129 corrections=data.get("corrections", []),
130 should_close=data.get("should_close", False),
131 close_reason=data.get("close_reason"),
132 close_status=data.get("close_status"),
133 was_blocked=data.get("was_blocked", False),
134 interrupted=data.get("interrupted", False),
135 )
138@dataclass
139class SprintWorkerContext:
140 """Sprint worker identity injected into guillotine continuation prompts.
142 Tells a fresh Option J session which single issue it must complete
143 and that it must exit immediately after — preventing the deadlock where
144 a fresh session processes multiple visible issues and blocks on "What next?".
146 Attributes:
147 issue_id: The single issue this worker is responsible for (e.g. "FEAT-025")
148 branch: Git branch for this worker (main or worktree branch)
149 """
151 issue_id: str
152 branch: str
154 def to_dict(self) -> dict[str, Any]:
155 """Convert to dictionary for JSON serialization."""
156 return {"issue_id": self.issue_id, "branch": self.branch}
159class MergeStatus(Enum):
160 """Status of a merge operation."""
162 PENDING = "pending"
163 IN_PROGRESS = "in_progress"
164 SUCCESS = "success"
165 CONFLICT = "conflict"
166 FAILED = "failed"
167 RETRYING = "retrying"
170class WorkerStage(Enum):
171 """Processing stage of a worker.
173 Stages progress in order:
174 - SETUP: Creating git worktree and copying .claude/ directory
175 - VALIDATING: Running ready-issue command
176 - IMPLEMENTING: Running manage-issue command
177 - VERIFYING: Checking work was done and updating branch base
178 - MERGING: Awaiting merge coordination
179 - COMPLETED: Successfully finished
180 - FAILED: Failed at some stage
181 - INTERRUPTED: Interrupted during shutdown
182 """
184 SETUP = "setup"
185 VALIDATING = "validating"
186 IMPLEMENTING = "implementing"
187 VERIFYING = "verifying"
188 MERGING = "merging"
189 COMPLETED = "completed"
190 FAILED = "failed"
191 INTERRUPTED = "interrupted"
194@dataclass
195class MergeRequest:
196 """Request to merge a completed worker's changes.
198 Attributes:
199 worker_result: The result from the worker
200 status: Current merge status
201 retry_count: Number of merge/rebase attempts
202 error: Error message if merge failed
203 queued_at: When the merge was requested
204 """
206 worker_result: WorkerResult
207 status: MergeStatus = MergeStatus.PENDING
208 retry_count: int = 0
209 error: str | None = None
210 queued_at: float = field(default_factory=time.time)
212 def to_dict(self) -> dict[str, Any]:
213 """Convert to dictionary for JSON serialization."""
214 return {
215 "worker_result": self.worker_result.to_dict(),
216 "status": self.status.value,
217 "retry_count": self.retry_count,
218 "error": self.error,
219 "queued_at": self.queued_at,
220 }
223@dataclass
224class OrchestratorState:
225 """Persistent state for the parallel orchestrator.
227 Enables resume capability after interruption.
229 Attributes:
230 in_progress_issues: Issues currently being processed by workers
231 completed_issues: Successfully completed issue IDs
232 failed_issues: Mapping of issue ID to failure reason
233 pending_merges: Issues awaiting merge
234 timing: Per-issue timing breakdown
235 corrections: Mapping of issue ID to list of corrections made (for pattern analysis)
236 started_at: When orchestration started
237 last_checkpoint: Last state save timestamp
238 """
240 in_progress_issues: list[str] = field(default_factory=list)
241 completed_issues: list[str] = field(default_factory=list)
242 failed_issues: dict[str, str] = field(default_factory=dict)
243 pending_merges: list[str] = field(default_factory=list)
244 timing: dict[str, dict[str, float]] = field(default_factory=dict)
245 corrections: dict[str, list[str]] = field(default_factory=dict)
246 started_at: str = ""
247 last_checkpoint: str = ""
249 def to_dict(self) -> dict[str, Any]:
250 """Convert state to dictionary for JSON serialization."""
251 return {
252 "in_progress_issues": self.in_progress_issues,
253 "completed_issues": self.completed_issues,
254 "failed_issues": self.failed_issues,
255 "pending_merges": self.pending_merges,
256 "timing": self.timing,
257 "corrections": self.corrections,
258 "started_at": self.started_at,
259 "last_checkpoint": self.last_checkpoint,
260 }
262 @classmethod
263 def from_dict(cls, data: dict[str, Any]) -> OrchestratorState:
264 """Create state from dictionary (JSON deserialization)."""
265 return cls(
266 in_progress_issues=data.get("in_progress_issues", []),
267 completed_issues=data.get("completed_issues", []),
268 failed_issues=data.get("failed_issues", {}),
269 pending_merges=data.get("pending_merges", []),
270 timing=data.get("timing", {}),
271 corrections=data.get("corrections", {}),
272 started_at=data.get("started_at", ""),
273 last_checkpoint=data.get("last_checkpoint", ""),
274 )
277@dataclass
278class PendingWorktreeInfo:
279 """Information about a pending worktree from a previous run.
281 Attributes:
282 worktree_path: Path to the worktree directory
283 branch_name: Git branch name (parallel/<issue-id>-<timestamp>)
284 issue_id: Extracted issue ID (e.g., BUG-045)
285 commits_ahead: Number of commits ahead of main
286 has_uncommitted_changes: Whether there are uncommitted changes
287 changed_files: List of files with uncommitted changes
288 """
290 worktree_path: Path
291 branch_name: str
292 issue_id: str
293 commits_ahead: int
294 has_uncommitted_changes: bool
295 changed_files: list[str] = field(default_factory=list)
297 @property
298 def has_pending_work(self) -> bool:
299 """Return True if this worktree has work that could be merged."""
300 return self.commits_ahead > 0 or self.has_uncommitted_changes
303@dataclass
304class ParallelConfig:
305 """Configuration for the parallel issue manager.
307 Supports configurable command templates for different project setups.
308 Commands use placeholders: {{issue_id}}, {{issue_type}}, {{action}}
310 Attributes:
311 max_workers: Number of parallel workers (default: 2)
312 p0_sequential: Process P0 issues sequentially (default: True)
313 merge_interval: Seconds between merge attempts (default: 30.0)
314 worktree_base: Base directory for git worktrees
315 state_file: Path to state persistence file
316 max_merge_retries: Maximum rebase attempts before giving up (default: 2)
317 priority_filter: Which priority levels to process
318 max_issues: Maximum issues to process (0 = unlimited)
319 dry_run: Preview mode without actual processing
320 timeout_per_issue: Timeout in seconds for each issue (default: 7200)
321 orchestrator_timeout: Timeout for waiting on workers (default: 0 = auto)
322 stream_subprocess_output: Whether to stream subprocess output
323 show_model: Make API call to verify and display model on worktree setup
324 command_prefix: Prefix for slash commands (default: "/ll:")
325 ready_command: Template for ready-issue command
326 manage_command: Template for manage-issue command
327 only_ids: If provided, only process these issue IDs
328 skip_ids: Issue IDs to skip (in addition to completed/failed)
329 merge_pending: Attempt to merge pending worktrees from previous runs
330 clean_start: Remove all worktrees without checking for pending work
331 ignore_pending: Report pending work but continue without merging
332 """
334 max_workers: int = 2
335 p0_sequential: bool = True
336 merge_interval: float = 30.0
337 worktree_base: Path = field(default_factory=lambda: Path(".worktrees"))
338 state_file: Path = field(default_factory=lambda: Path(".parallel-manage-state.json"))
339 max_merge_retries: int = 2
340 priority_filter: list[str] = field(default_factory=lambda: ["P0", "P1", "P2", "P3", "P4", "P5"])
341 max_issues: int = 0
342 dry_run: bool = False
343 timeout_per_issue: int = 3600
344 idle_timeout_per_issue: int = 0 # Kill if no output for N seconds (0 to disable)
345 orchestrator_timeout: int = 0 # 0 = use timeout_per_issue * max_workers
346 stream_subprocess_output: bool = False
347 show_model: bool = False # Make API call to verify model on worktree setup
348 # Configurable command templates
349 command_prefix: str = "/ll:"
350 ready_command: str = "ready-issue {{issue_id}}"
351 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}"
352 decide_command: str = "decide-issue {{issue_id}}"
353 # Issue ID and label filters
354 only_ids: set[str] | None = None
355 skip_ids: set[str] | None = None
356 type_prefixes: set[str] | None = None
357 label_filter: set[str] | None = None
358 # Validation settings
359 require_code_changes: bool = True # If False, allow changes to only excluded dirs
360 use_feature_branches: bool = (
361 False # If True, create feature/<id>-<slug> branch instead of parallel/
362 )
363 push_feature_branches: bool = False # If True, push feature branch to remote_name after success
364 open_pr_for_feature_branches: bool = False # If True, open a draft PR after push via gh CLI
365 # Additional files to copy from main repo to worktrees
366 # Note: .claude/ directory is always copied automatically (see worker_pool.py)
367 worktree_copy_files: list[str] = field(
368 default_factory=lambda: [".claude/settings.local.json", ".env"]
369 )
370 # Pending worktree handling flags
371 merge_pending: bool = False # Attempt to merge pending worktrees
372 clean_start: bool = False # Remove all worktrees without checking
373 ignore_pending: bool = False # Report pending work but continue
374 # Overlap detection settings (ENH-143)
375 overlap_detection: bool = False # Enable pre-flight overlap detection
376 serialize_overlapping: bool = True # If True, defer overlapping issues; if False, just warn
377 # Base branch for rebase/merge operations (auto-detected at startup)
378 base_branch: str = "main"
379 # Git remote name for fetch/pull operations
380 remote_name: str = "origin"
382 def get_ready_command(self, issue_id: str) -> str:
383 """Build the ready-issue command string.
385 Args:
386 issue_id: Issue identifier
388 Returns:
389 Complete command string
390 """
391 cmd = self.ready_command.replace("{{issue_id}}", issue_id)
392 return f"{self.command_prefix}{cmd}"
394 def get_manage_command(self, issue_type: str, action: str, issue_id: str) -> str:
395 """Build the manage-issue command string.
397 Args:
398 issue_type: Type of issue (bug, feature, enhancement)
399 action: Action to perform (fix, implement, improve)
400 issue_id: Issue identifier
402 Returns:
403 Complete command string
404 """
405 cmd = (
406 self.manage_command.replace("{{issue_type}}", issue_type)
407 .replace("{{action}}", action)
408 .replace("{{issue_id}}", issue_id)
409 )
410 return f"{self.command_prefix}{cmd}"
412 def get_decide_command(self, issue_id: str) -> str:
413 """Build the decide-issue command string.
415 Args:
416 issue_id: Issue identifier
418 Returns:
419 Complete command string
420 """
421 cmd = self.decide_command.replace("{{issue_id}}", issue_id)
422 return f"{self.command_prefix}{cmd}"
424 def to_dict(self) -> dict[str, Any]:
425 """Convert to dictionary for JSON serialization."""
426 return {
427 "max_workers": self.max_workers,
428 "p0_sequential": self.p0_sequential,
429 "merge_interval": self.merge_interval,
430 "worktree_base": str(self.worktree_base),
431 "state_file": str(self.state_file),
432 "max_merge_retries": self.max_merge_retries,
433 "priority_filter": self.priority_filter,
434 "max_issues": self.max_issues,
435 "dry_run": self.dry_run,
436 "timeout_per_issue": self.timeout_per_issue,
437 "idle_timeout_per_issue": self.idle_timeout_per_issue,
438 "orchestrator_timeout": self.orchestrator_timeout,
439 "stream_subprocess_output": self.stream_subprocess_output,
440 "show_model": self.show_model,
441 "command_prefix": self.command_prefix,
442 "ready_command": self.ready_command,
443 "manage_command": self.manage_command,
444 "decide_command": self.decide_command,
445 "only_ids": list(self.only_ids) if self.only_ids else None,
446 "skip_ids": list(self.skip_ids) if self.skip_ids else None,
447 "type_prefixes": list(self.type_prefixes) if self.type_prefixes else None,
448 "label_filter": list(self.label_filter) if self.label_filter else None,
449 "require_code_changes": self.require_code_changes,
450 "use_feature_branches": self.use_feature_branches,
451 "push_feature_branches": self.push_feature_branches,
452 "open_pr_for_feature_branches": self.open_pr_for_feature_branches,
453 "merge_pending": self.merge_pending,
454 "clean_start": self.clean_start,
455 "ignore_pending": self.ignore_pending,
456 "overlap_detection": self.overlap_detection,
457 "serialize_overlapping": self.serialize_overlapping,
458 "base_branch": self.base_branch,
459 "remote_name": self.remote_name,
460 }
462 @classmethod
463 def from_dict(cls, data: dict[str, Any]) -> ParallelConfig:
464 """Create from dictionary (JSON deserialization)."""
465 only_ids_data = data.get("only_ids")
466 skip_ids_data = data.get("skip_ids")
467 type_prefixes_data = data.get("type_prefixes")
468 label_filter_data = data.get("label_filter")
469 return cls(
470 max_workers=data.get("max_workers", 2),
471 p0_sequential=data.get("p0_sequential", True),
472 merge_interval=data.get("merge_interval", 30.0),
473 worktree_base=Path(data.get("worktree_base", ".worktrees")),
474 state_file=Path(data.get("state_file", ".parallel-manage-state.json")),
475 max_merge_retries=data.get("max_merge_retries", 2),
476 priority_filter=data.get("priority_filter", ["P0", "P1", "P2", "P3", "P4", "P5"]),
477 max_issues=data.get("max_issues", 0),
478 dry_run=data.get("dry_run", False),
479 timeout_per_issue=data.get("timeout_per_issue", 7200),
480 idle_timeout_per_issue=data.get("idle_timeout_per_issue", 0),
481 orchestrator_timeout=data.get("orchestrator_timeout", 0),
482 stream_subprocess_output=data.get("stream_subprocess_output", False),
483 show_model=data.get("show_model", False),
484 command_prefix=data.get("command_prefix", "/ll:"),
485 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"),
486 manage_command=data.get(
487 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}"
488 ),
489 decide_command=data.get("decide_command", "decide-issue {{issue_id}}"),
490 only_ids=set(only_ids_data) if only_ids_data else None,
491 skip_ids=set(skip_ids_data) if skip_ids_data else None,
492 type_prefixes=set(type_prefixes_data) if type_prefixes_data else None,
493 label_filter=set(label_filter_data) if label_filter_data else None,
494 require_code_changes=data.get("require_code_changes", True),
495 use_feature_branches=data.get("use_feature_branches", False),
496 push_feature_branches=data.get("push_feature_branches", False),
497 open_pr_for_feature_branches=data.get("open_pr_for_feature_branches", False),
498 merge_pending=data.get("merge_pending", False),
499 clean_start=data.get("clean_start", False),
500 ignore_pending=data.get("ignore_pending", False),
501 overlap_detection=data.get("overlap_detection", False),
502 serialize_overlapping=data.get("serialize_overlapping", True),
503 base_branch=data.get("base_branch", "main"),
504 remote_name=data.get("remote_name", "origin"),
505 )