Coverage for little_loops / parallel / types.py: 86%
138 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -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 )
138class MergeStatus(Enum):
139 """Status of a merge operation."""
141 PENDING = "pending"
142 IN_PROGRESS = "in_progress"
143 SUCCESS = "success"
144 CONFLICT = "conflict"
145 FAILED = "failed"
146 RETRYING = "retrying"
149class WorkerStage(Enum):
150 """Processing stage of a worker.
152 Stages progress in order:
153 - SETUP: Creating git worktree and copying .claude/ directory
154 - VALIDATING: Running ready-issue command
155 - IMPLEMENTING: Running manage-issue command
156 - VERIFYING: Checking work was done and updating branch base
157 - MERGING: Awaiting merge coordination
158 - COMPLETED: Successfully finished
159 - FAILED: Failed at some stage
160 - INTERRUPTED: Interrupted during shutdown
161 """
163 SETUP = "setup"
164 VALIDATING = "validating"
165 IMPLEMENTING = "implementing"
166 VERIFYING = "verifying"
167 MERGING = "merging"
168 COMPLETED = "completed"
169 FAILED = "failed"
170 INTERRUPTED = "interrupted"
173@dataclass
174class MergeRequest:
175 """Request to merge a completed worker's changes.
177 Attributes:
178 worker_result: The result from the worker
179 status: Current merge status
180 retry_count: Number of merge/rebase attempts
181 error: Error message if merge failed
182 queued_at: When the merge was requested
183 """
185 worker_result: WorkerResult
186 status: MergeStatus = MergeStatus.PENDING
187 retry_count: int = 0
188 error: str | None = None
189 queued_at: float = field(default_factory=time.time)
191 def to_dict(self) -> dict[str, Any]:
192 """Convert to dictionary for JSON serialization."""
193 return {
194 "worker_result": self.worker_result.to_dict(),
195 "status": self.status.value,
196 "retry_count": self.retry_count,
197 "error": self.error,
198 "queued_at": self.queued_at,
199 }
202@dataclass
203class OrchestratorState:
204 """Persistent state for the parallel orchestrator.
206 Enables resume capability after interruption.
208 Attributes:
209 in_progress_issues: Issues currently being processed by workers
210 completed_issues: Successfully completed issue IDs
211 failed_issues: Mapping of issue ID to failure reason
212 pending_merges: Issues awaiting merge
213 timing: Per-issue timing breakdown
214 corrections: Mapping of issue ID to list of corrections made (for pattern analysis)
215 started_at: When orchestration started
216 last_checkpoint: Last state save timestamp
217 """
219 in_progress_issues: list[str] = field(default_factory=list)
220 completed_issues: list[str] = field(default_factory=list)
221 failed_issues: dict[str, str] = field(default_factory=dict)
222 pending_merges: list[str] = field(default_factory=list)
223 timing: dict[str, dict[str, float]] = field(default_factory=dict)
224 corrections: dict[str, list[str]] = field(default_factory=dict)
225 started_at: str = ""
226 last_checkpoint: str = ""
228 def to_dict(self) -> dict[str, Any]:
229 """Convert state to dictionary for JSON serialization."""
230 return {
231 "in_progress_issues": self.in_progress_issues,
232 "completed_issues": self.completed_issues,
233 "failed_issues": self.failed_issues,
234 "pending_merges": self.pending_merges,
235 "timing": self.timing,
236 "corrections": self.corrections,
237 "started_at": self.started_at,
238 "last_checkpoint": self.last_checkpoint,
239 }
241 @classmethod
242 def from_dict(cls, data: dict[str, Any]) -> OrchestratorState:
243 """Create state from dictionary (JSON deserialization)."""
244 return cls(
245 in_progress_issues=data.get("in_progress_issues", []),
246 completed_issues=data.get("completed_issues", []),
247 failed_issues=data.get("failed_issues", {}),
248 pending_merges=data.get("pending_merges", []),
249 timing=data.get("timing", {}),
250 corrections=data.get("corrections", {}),
251 started_at=data.get("started_at", ""),
252 last_checkpoint=data.get("last_checkpoint", ""),
253 )
256@dataclass
257class PendingWorktreeInfo:
258 """Information about a pending worktree from a previous run.
260 Attributes:
261 worktree_path: Path to the worktree directory
262 branch_name: Git branch name (parallel/<issue-id>-<timestamp>)
263 issue_id: Extracted issue ID (e.g., BUG-045)
264 commits_ahead: Number of commits ahead of main
265 has_uncommitted_changes: Whether there are uncommitted changes
266 changed_files: List of files with uncommitted changes
267 """
269 worktree_path: Path
270 branch_name: str
271 issue_id: str
272 commits_ahead: int
273 has_uncommitted_changes: bool
274 changed_files: list[str] = field(default_factory=list)
276 @property
277 def has_pending_work(self) -> bool:
278 """Return True if this worktree has work that could be merged."""
279 return self.commits_ahead > 0 or self.has_uncommitted_changes
282@dataclass
283class ParallelConfig:
284 """Configuration for the parallel issue manager.
286 Supports configurable command templates for different project setups.
287 Commands use placeholders: {{issue_id}}, {{issue_type}}, {{action}}
289 Attributes:
290 max_workers: Number of parallel workers (default: 2)
291 p0_sequential: Process P0 issues sequentially (default: True)
292 merge_interval: Seconds between merge attempts (default: 30.0)
293 worktree_base: Base directory for git worktrees
294 state_file: Path to state persistence file
295 max_merge_retries: Maximum rebase attempts before giving up (default: 2)
296 priority_filter: Which priority levels to process
297 max_issues: Maximum issues to process (0 = unlimited)
298 dry_run: Preview mode without actual processing
299 timeout_per_issue: Timeout in seconds for each issue (default: 7200)
300 orchestrator_timeout: Timeout for waiting on workers (default: 0 = auto)
301 stream_subprocess_output: Whether to stream subprocess output
302 show_model: Make API call to verify and display model on worktree setup
303 command_prefix: Prefix for slash commands (default: "/ll:")
304 ready_command: Template for ready-issue command
305 manage_command: Template for manage-issue command
306 only_ids: If provided, only process these issue IDs
307 skip_ids: Issue IDs to skip (in addition to completed/failed)
308 merge_pending: Attempt to merge pending worktrees from previous runs
309 clean_start: Remove all worktrees without checking for pending work
310 ignore_pending: Report pending work but continue without merging
311 """
313 max_workers: int = 2
314 p0_sequential: bool = True
315 merge_interval: float = 30.0
316 worktree_base: Path = field(default_factory=lambda: Path(".worktrees"))
317 state_file: Path = field(default_factory=lambda: Path(".parallel-manage-state.json"))
318 max_merge_retries: int = 2
319 priority_filter: list[str] = field(default_factory=lambda: ["P0", "P1", "P2", "P3", "P4", "P5"])
320 max_issues: int = 0
321 dry_run: bool = False
322 timeout_per_issue: int = 3600
323 idle_timeout_per_issue: int = 0 # Kill if no output for N seconds (0 to disable)
324 orchestrator_timeout: int = 0 # 0 = use timeout_per_issue * max_workers
325 stream_subprocess_output: bool = False
326 show_model: bool = False # Make API call to verify model on worktree setup
327 # Configurable command templates
328 command_prefix: str = "/ll:"
329 ready_command: str = "ready-issue {{issue_id}}"
330 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}"
331 # Issue ID filters
332 only_ids: set[str] | None = None
333 skip_ids: set[str] | None = None
334 type_prefixes: set[str] | None = None
335 # Validation settings
336 require_code_changes: bool = True # If False, allow changes to only excluded dirs
337 use_feature_branches: bool = (
338 False # If True, create feature/<id>-<slug> branch instead of parallel/
339 )
340 # Additional files to copy from main repo to worktrees
341 # Note: .claude/ directory is always copied automatically (see worker_pool.py)
342 worktree_copy_files: list[str] = field(
343 default_factory=lambda: [".claude/settings.local.json", ".env"]
344 )
345 # Pending worktree handling flags
346 merge_pending: bool = False # Attempt to merge pending worktrees
347 clean_start: bool = False # Remove all worktrees without checking
348 ignore_pending: bool = False # Report pending work but continue
349 # Overlap detection settings (ENH-143)
350 overlap_detection: bool = False # Enable pre-flight overlap detection
351 serialize_overlapping: bool = True # If True, defer overlapping issues; if False, just warn
352 # Base branch for rebase/merge operations (auto-detected at startup)
353 base_branch: str = "main"
354 # Git remote name for fetch/pull operations
355 remote_name: str = "origin"
357 def get_ready_command(self, issue_id: str) -> str:
358 """Build the ready-issue command string.
360 Args:
361 issue_id: Issue identifier
363 Returns:
364 Complete command string
365 """
366 cmd = self.ready_command.replace("{{issue_id}}", issue_id)
367 return f"{self.command_prefix}{cmd}"
369 def get_manage_command(self, issue_type: str, action: str, issue_id: str) -> str:
370 """Build the manage-issue command string.
372 Args:
373 issue_type: Type of issue (bug, feature, enhancement)
374 action: Action to perform (fix, implement, improve)
375 issue_id: Issue identifier
377 Returns:
378 Complete command string
379 """
380 cmd = (
381 self.manage_command.replace("{{issue_type}}", issue_type)
382 .replace("{{action}}", action)
383 .replace("{{issue_id}}", issue_id)
384 )
385 return f"{self.command_prefix}{cmd}"
387 def to_dict(self) -> dict[str, Any]:
388 """Convert to dictionary for JSON serialization."""
389 return {
390 "max_workers": self.max_workers,
391 "p0_sequential": self.p0_sequential,
392 "merge_interval": self.merge_interval,
393 "worktree_base": str(self.worktree_base),
394 "state_file": str(self.state_file),
395 "max_merge_retries": self.max_merge_retries,
396 "priority_filter": self.priority_filter,
397 "max_issues": self.max_issues,
398 "dry_run": self.dry_run,
399 "timeout_per_issue": self.timeout_per_issue,
400 "idle_timeout_per_issue": self.idle_timeout_per_issue,
401 "orchestrator_timeout": self.orchestrator_timeout,
402 "stream_subprocess_output": self.stream_subprocess_output,
403 "show_model": self.show_model,
404 "command_prefix": self.command_prefix,
405 "ready_command": self.ready_command,
406 "manage_command": self.manage_command,
407 "only_ids": list(self.only_ids) if self.only_ids else None,
408 "skip_ids": list(self.skip_ids) if self.skip_ids else None,
409 "type_prefixes": list(self.type_prefixes) if self.type_prefixes else None,
410 "require_code_changes": self.require_code_changes,
411 "use_feature_branches": self.use_feature_branches,
412 "merge_pending": self.merge_pending,
413 "clean_start": self.clean_start,
414 "ignore_pending": self.ignore_pending,
415 "overlap_detection": self.overlap_detection,
416 "serialize_overlapping": self.serialize_overlapping,
417 "base_branch": self.base_branch,
418 "remote_name": self.remote_name,
419 }
421 @classmethod
422 def from_dict(cls, data: dict[str, Any]) -> ParallelConfig:
423 """Create from dictionary (JSON deserialization)."""
424 only_ids_data = data.get("only_ids")
425 skip_ids_data = data.get("skip_ids")
426 type_prefixes_data = data.get("type_prefixes")
427 return cls(
428 max_workers=data.get("max_workers", 2),
429 p0_sequential=data.get("p0_sequential", True),
430 merge_interval=data.get("merge_interval", 30.0),
431 worktree_base=Path(data.get("worktree_base", ".worktrees")),
432 state_file=Path(data.get("state_file", ".parallel-manage-state.json")),
433 max_merge_retries=data.get("max_merge_retries", 2),
434 priority_filter=data.get("priority_filter", ["P0", "P1", "P2", "P3", "P4", "P5"]),
435 max_issues=data.get("max_issues", 0),
436 dry_run=data.get("dry_run", False),
437 timeout_per_issue=data.get("timeout_per_issue", 7200),
438 idle_timeout_per_issue=data.get("idle_timeout_per_issue", 0),
439 orchestrator_timeout=data.get("orchestrator_timeout", 0),
440 stream_subprocess_output=data.get("stream_subprocess_output", False),
441 show_model=data.get("show_model", False),
442 command_prefix=data.get("command_prefix", "/ll:"),
443 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"),
444 manage_command=data.get(
445 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}"
446 ),
447 only_ids=set(only_ids_data) if only_ids_data else None,
448 skip_ids=set(skip_ids_data) if skip_ids_data else None,
449 type_prefixes=set(type_prefixes_data) if type_prefixes_data else None,
450 require_code_changes=data.get("require_code_changes", True),
451 use_feature_branches=data.get("use_feature_branches", False),
452 merge_pending=data.get("merge_pending", False),
453 clean_start=data.get("clean_start", False),
454 ignore_pending=data.get("ignore_pending", False),
455 overlap_detection=data.get("overlap_detection", False),
456 serialize_overlapping=data.get("serialize_overlapping", True),
457 base_branch=data.get("base_branch", "main"),
458 remote_name=data.get("remote_name", "origin"),
459 )