Coverage for little_loops / parallel / types.py: 85%
154 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -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 - PROVING: Running proof-first-task assumption-firewall gate (ENH-2219)
177 - IMPLEMENTING: Running manage-issue command
178 - VERIFYING: Checking work was done and updating branch base
179 - MERGING: Awaiting merge coordination
180 - COMPLETED: Successfully finished
181 - FAILED: Failed at some stage
182 - INTERRUPTED: Interrupted during shutdown
183 """
185 SETUP = "setup"
186 VALIDATING = "validating"
187 PROVING = "proving"
188 IMPLEMENTING = "implementing"
189 VERIFYING = "verifying"
190 MERGING = "merging"
191 COMPLETED = "completed"
192 FAILED = "failed"
193 INTERRUPTED = "interrupted"
196@dataclass
197class MergeRequest:
198 """Request to merge a completed worker's changes.
200 Attributes:
201 worker_result: The result from the worker
202 status: Current merge status
203 retry_count: Number of merge/rebase attempts
204 error: Error message if merge failed
205 queued_at: When the merge was requested
206 """
208 worker_result: WorkerResult
209 status: MergeStatus = MergeStatus.PENDING
210 retry_count: int = 0
211 error: str | None = None
212 queued_at: float = field(default_factory=time.time)
214 def to_dict(self) -> dict[str, Any]:
215 """Convert to dictionary for JSON serialization."""
216 return {
217 "worker_result": self.worker_result.to_dict(),
218 "status": self.status.value,
219 "retry_count": self.retry_count,
220 "error": self.error,
221 "queued_at": self.queued_at,
222 }
225@dataclass
226class OrchestratorState:
227 """Persistent state for the parallel orchestrator.
229 Enables resume capability after interruption.
231 Attributes:
232 in_progress_issues: Issues currently being processed by workers
233 completed_issues: Successfully completed issue IDs
234 failed_issues: Mapping of issue ID to failure reason
235 pending_merges: Issues awaiting merge
236 timing: Per-issue timing breakdown
237 corrections: Mapping of issue ID to list of corrections made (for pattern analysis)
238 started_at: When orchestration started
239 last_checkpoint: Last state save timestamp
240 """
242 in_progress_issues: list[str] = field(default_factory=list)
243 completed_issues: list[str] = field(default_factory=list)
244 failed_issues: dict[str, str] = field(default_factory=dict)
245 pending_merges: list[str] = field(default_factory=list)
246 timing: dict[str, dict[str, float]] = field(default_factory=dict)
247 corrections: dict[str, list[str]] = field(default_factory=dict)
248 started_at: str = ""
249 last_checkpoint: str = ""
251 def to_dict(self) -> dict[str, Any]:
252 """Convert state to dictionary for JSON serialization."""
253 return {
254 "in_progress_issues": self.in_progress_issues,
255 "completed_issues": self.completed_issues,
256 "failed_issues": self.failed_issues,
257 "pending_merges": self.pending_merges,
258 "timing": self.timing,
259 "corrections": self.corrections,
260 "started_at": self.started_at,
261 "last_checkpoint": self.last_checkpoint,
262 }
264 @classmethod
265 def from_dict(cls, data: dict[str, Any]) -> OrchestratorState:
266 """Create state from dictionary (JSON deserialization)."""
267 return cls(
268 in_progress_issues=data.get("in_progress_issues", []),
269 completed_issues=data.get("completed_issues", []),
270 failed_issues=data.get("failed_issues", {}),
271 pending_merges=data.get("pending_merges", []),
272 timing=data.get("timing", {}),
273 corrections=data.get("corrections", {}),
274 started_at=data.get("started_at", ""),
275 last_checkpoint=data.get("last_checkpoint", ""),
276 )
279@dataclass
280class PendingWorktreeInfo:
281 """Information about a pending worktree from a previous run.
283 Attributes:
284 worktree_path: Path to the worktree directory
285 branch_name: Git branch name (parallel/<issue-id>-<timestamp>)
286 issue_id: Extracted issue ID (e.g., BUG-045)
287 commits_ahead: Number of commits ahead of main
288 has_uncommitted_changes: Whether there are uncommitted changes
289 changed_files: List of files with uncommitted changes
290 """
292 worktree_path: Path
293 branch_name: str
294 issue_id: str
295 commits_ahead: int
296 has_uncommitted_changes: bool
297 changed_files: list[str] = field(default_factory=list)
299 @property
300 def has_pending_work(self) -> bool:
301 """Return True if this worktree has work that could be merged."""
302 return self.commits_ahead > 0 or self.has_uncommitted_changes
305@dataclass
306class ParallelConfig:
307 """Configuration for the parallel issue manager.
309 Supports configurable command templates for different project setups.
310 Commands use placeholders: {{issue_id}}, {{issue_type}}, {{action}}
312 Attributes:
313 max_workers: Number of parallel workers (default: 2)
314 p0_sequential: Process P0 issues sequentially (default: True)
315 merge_interval: Seconds between merge attempts (default: 30.0)
316 worktree_base: Base directory for git worktrees
317 state_file: Path to state persistence file
318 max_merge_retries: Maximum rebase attempts before giving up (default: 2)
319 priority_filter: Which priority levels to process
320 max_issues: Maximum issues to process (0 = unlimited)
321 dry_run: Preview mode without actual processing
322 timeout_per_issue: Timeout in seconds for each issue (default: 7200)
323 orchestrator_timeout: Timeout for waiting on workers (default: 0 = auto)
324 stream_subprocess_output: Whether to stream subprocess output
325 show_model: Make API call to verify and display model on worktree setup
326 command_prefix: Prefix for slash commands (default: "/ll:")
327 ready_command: Template for ready-issue command
328 manage_command: Template for manage-issue command
329 only_ids: If provided, only process these issue IDs
330 skip_ids: Issue IDs to skip (in addition to completed/failed)
331 merge_pending: Attempt to merge pending worktrees from previous runs
332 clean_start: Remove all worktrees without checking for pending work
333 ignore_pending: Report pending work but continue without merging
334 """
336 max_workers: int = 2
337 p0_sequential: bool = True
338 merge_interval: float = 30.0
339 worktree_base: Path = field(default_factory=lambda: Path(".worktrees"))
340 state_file: Path = field(default_factory=lambda: Path(".parallel-manage-state.json"))
341 max_merge_retries: int = 2
342 priority_filter: list[str] = field(default_factory=lambda: ["P0", "P1", "P2", "P3", "P4", "P5"])
343 max_issues: int = 0
344 dry_run: bool = False
345 timeout_per_issue: int = 3600
346 idle_timeout_per_issue: int = 0 # Kill if no output for N seconds (0 to disable)
347 orchestrator_timeout: int = 0 # 0 = use timeout_per_issue * max_workers
348 stream_subprocess_output: bool = False
349 show_model: bool = False # Make API call to verify model on worktree setup
350 # Configurable command templates
351 command_prefix: str = "/ll:"
352 ready_command: str = "ready-issue {{issue_id}}"
353 manage_command: str = "manage-issue {{issue_type}} {{action}} {{issue_id}}"
354 decide_command: str = "decide-issue {{issue_id}}"
355 # Issue ID and label filters
356 only_ids: set[str] | None = None
357 skip_ids: set[str] | None = None
358 type_prefixes: set[str] | None = None
359 label_filter: set[str] | None = None
360 # Validation settings
361 require_code_changes: bool = True # If False, allow changes to only excluded dirs
362 use_feature_branches: bool = (
363 False # If True, create feature/<id>-<slug> branch instead of parallel/
364 )
365 push_feature_branches: bool = False # If True, push feature branch to remote_name after success
366 open_pr_for_feature_branches: bool = False # If True, open a draft PR after push via gh CLI
367 # Additional files to copy from main repo to worktrees
368 # Note: .claude/ directory is always copied automatically (see worker_pool.py)
369 worktree_copy_files: list[str] = field(
370 default_factory=lambda: [".claude/settings.local.json", ".env"]
371 )
372 # Pending worktree handling flags
373 merge_pending: bool = False # Attempt to merge pending worktrees
374 clean_start: bool = False # Remove all worktrees without checking
375 ignore_pending: bool = False # Report pending work but continue
376 # Overlap detection settings (ENH-143)
377 overlap_detection: bool = False # Enable pre-flight overlap detection
378 serialize_overlapping: bool = True # If True, defer overlapping issues; if False, just warn
379 # Learning test gate (ENH-2219)
380 skip_learning_gate: bool = False # Bypass per-worktree proof-first-task gate
381 # Base branch for rebase/merge operations (auto-detected at startup)
382 base_branch: str = "main"
383 # Git remote name for fetch/pull operations
384 remote_name: str = "origin"
386 def get_ready_command(self, issue_id: str) -> str:
387 """Build the ready-issue command string.
389 Args:
390 issue_id: Issue identifier
392 Returns:
393 Complete command string
394 """
395 cmd = self.ready_command.replace("{{issue_id}}", issue_id)
396 return f"{self.command_prefix}{cmd}"
398 def get_manage_command(self, issue_type: str, action: str, issue_id: str) -> str:
399 """Build the manage-issue command string.
401 Args:
402 issue_type: Type of issue (bug, feature, enhancement)
403 action: Action to perform (fix, implement, improve)
404 issue_id: Issue identifier
406 Returns:
407 Complete command string
408 """
409 cmd = (
410 self.manage_command.replace("{{issue_type}}", issue_type)
411 .replace("{{action}}", action)
412 .replace("{{issue_id}}", issue_id)
413 )
414 return f"{self.command_prefix}{cmd}"
416 def get_decide_command(self, issue_id: str) -> str:
417 """Build the decide-issue command string.
419 Args:
420 issue_id: Issue identifier
422 Returns:
423 Complete command string
424 """
425 cmd = self.decide_command.replace("{{issue_id}}", issue_id)
426 return f"{self.command_prefix}{cmd}"
428 def to_dict(self) -> dict[str, Any]:
429 """Convert to dictionary for JSON serialization."""
430 return {
431 "max_workers": self.max_workers,
432 "p0_sequential": self.p0_sequential,
433 "merge_interval": self.merge_interval,
434 "worktree_base": str(self.worktree_base),
435 "state_file": str(self.state_file),
436 "max_merge_retries": self.max_merge_retries,
437 "priority_filter": self.priority_filter,
438 "max_issues": self.max_issues,
439 "dry_run": self.dry_run,
440 "timeout_per_issue": self.timeout_per_issue,
441 "idle_timeout_per_issue": self.idle_timeout_per_issue,
442 "orchestrator_timeout": self.orchestrator_timeout,
443 "stream_subprocess_output": self.stream_subprocess_output,
444 "show_model": self.show_model,
445 "command_prefix": self.command_prefix,
446 "ready_command": self.ready_command,
447 "manage_command": self.manage_command,
448 "decide_command": self.decide_command,
449 "only_ids": list(self.only_ids) if self.only_ids else None,
450 "skip_ids": list(self.skip_ids) if self.skip_ids else None,
451 "type_prefixes": list(self.type_prefixes) if self.type_prefixes else None,
452 "label_filter": list(self.label_filter) if self.label_filter else None,
453 "require_code_changes": self.require_code_changes,
454 "use_feature_branches": self.use_feature_branches,
455 "push_feature_branches": self.push_feature_branches,
456 "open_pr_for_feature_branches": self.open_pr_for_feature_branches,
457 "merge_pending": self.merge_pending,
458 "clean_start": self.clean_start,
459 "ignore_pending": self.ignore_pending,
460 "overlap_detection": self.overlap_detection,
461 "serialize_overlapping": self.serialize_overlapping,
462 "skip_learning_gate": self.skip_learning_gate,
463 "base_branch": self.base_branch,
464 "remote_name": self.remote_name,
465 }
467 @classmethod
468 def from_dict(cls, data: dict[str, Any]) -> ParallelConfig:
469 """Create from dictionary (JSON deserialization)."""
470 only_ids_data = data.get("only_ids")
471 skip_ids_data = data.get("skip_ids")
472 type_prefixes_data = data.get("type_prefixes")
473 label_filter_data = data.get("label_filter")
474 return cls(
475 max_workers=data.get("max_workers", 2),
476 p0_sequential=data.get("p0_sequential", True),
477 merge_interval=data.get("merge_interval", 30.0),
478 worktree_base=Path(data.get("worktree_base", ".worktrees")),
479 state_file=Path(data.get("state_file", ".parallel-manage-state.json")),
480 max_merge_retries=data.get("max_merge_retries", 2),
481 priority_filter=data.get("priority_filter", ["P0", "P1", "P2", "P3", "P4", "P5"]),
482 max_issues=data.get("max_issues", 0),
483 dry_run=data.get("dry_run", False),
484 timeout_per_issue=data.get("timeout_per_issue", 7200),
485 idle_timeout_per_issue=data.get("idle_timeout_per_issue", 0),
486 orchestrator_timeout=data.get("orchestrator_timeout", 0),
487 stream_subprocess_output=data.get("stream_subprocess_output", False),
488 show_model=data.get("show_model", False),
489 command_prefix=data.get("command_prefix", "/ll:"),
490 ready_command=data.get("ready_command", "ready-issue {{issue_id}}"),
491 manage_command=data.get(
492 "manage_command", "manage-issue {{issue_type}} {{action}} {{issue_id}}"
493 ),
494 decide_command=data.get("decide_command", "decide-issue {{issue_id}}"),
495 only_ids=set(only_ids_data) if only_ids_data else None,
496 skip_ids=set(skip_ids_data) if skip_ids_data else None,
497 type_prefixes=set(type_prefixes_data) if type_prefixes_data else None,
498 label_filter=set(label_filter_data) if label_filter_data else None,
499 require_code_changes=data.get("require_code_changes", True),
500 use_feature_branches=data.get("use_feature_branches", False),
501 push_feature_branches=data.get("push_feature_branches", False),
502 open_pr_for_feature_branches=data.get("open_pr_for_feature_branches", False),
503 merge_pending=data.get("merge_pending", False),
504 clean_start=data.get("clean_start", False),
505 ignore_pending=data.get("ignore_pending", False),
506 overlap_detection=data.get("overlap_detection", False),
507 serialize_overlapping=data.get("serialize_overlapping", True),
508 skip_learning_gate=data.get("skip_learning_gate", False),
509 base_branch=data.get("base_branch", "main"),
510 remote_name=data.get("remote_name", "origin"),
511 )