Coverage for little_loops / dependency_graph.py: 15%
214 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:08 -0500
1"""Dependency graph for issue management.
3Constructs a directed acyclic graph (DAG) from issue dependencies,
4providing topological sorting and cycle detection.
5"""
7from __future__ import annotations
9import logging
10from collections import deque
11from dataclasses import dataclass, field
12from typing import TYPE_CHECKING
14if TYPE_CHECKING:
15 from little_loops.config import DependencyMappingConfig
16 from little_loops.issue_parser import IssueInfo
18logger = logging.getLogger(__name__)
21@dataclass
22class WaveContentionNote:
23 """Annotation for a wave that was split due to file overlap."""
25 contended_paths: list[str]
26 sub_wave_index: int
27 total_sub_waves: int
28 parent_wave_index: int = 0
29 has_unknown_hints: bool = False # True when split includes issues with no file/directory hints
32@dataclass
33class DependencyGraph:
34 """Directed acyclic graph of issue dependencies.
36 Builds a graph from issue dependencies where edges represent
37 "blocked by" relationships. Provides methods for:
38 - Topological sorting (dependency order)
39 - Cycle detection
40 - Ready issue queries (blockers resolved)
41 - Blocking issue queries
43 Attributes:
44 issues: Mapping from issue_id to IssueInfo
45 blocked_by: Mapping from issue_id to set of blocking issue_ids
46 blocks: Mapping from issue_id to set of blocked issue_ids
47 depends_on_edges: Mapping from issue_id to set of soft-prerequisite issue_ids
48 """
50 issues: dict[str, IssueInfo] = field(default_factory=dict)
51 blocked_by: dict[str, set[str]] = field(default_factory=dict)
52 blocks: dict[str, set[str]] = field(default_factory=dict)
53 depends_on_edges: dict[str, set[str]] = field(default_factory=dict)
55 @classmethod
56 def from_issues(
57 cls,
58 issues: list[IssueInfo],
59 completed_ids: set[str] | None = None,
60 all_known_ids: set[str] | None = None,
61 ) -> DependencyGraph:
62 """Build graph from list of issues.
64 Constructs a dependency graph where:
65 - Each issue is a node
66 - blocked_by relationships create edges from blockers to blocked issues
67 - Completed issues are treated as satisfied (not blocking)
68 - Missing issues are logged as warnings but don't block
70 Args:
71 issues: List of IssueInfo objects with blocked_by/blocks fields
72 completed_ids: Set of completed issue IDs (treated as resolved)
73 all_known_ids: Set of all issue IDs that exist on disk. When
74 provided, references to issues in this set are silently
75 skipped (not warned) even if they are not in the graph.
77 Returns:
78 Constructed DependencyGraph
80 Example:
81 >>> issues = [issue_a, issue_b, issue_c]
82 >>> completed = {"FEAT-001"} # Already done
83 >>> graph = DependencyGraph.from_issues(issues, completed)
84 """
85 completed = completed_ids or set()
86 graph = cls()
88 # Index all issues by ID
89 for issue in issues:
90 graph.issues[issue.issue_id] = issue
91 graph.blocked_by[issue.issue_id] = set()
92 graph.blocks[issue.issue_id] = set()
94 # Build dependency edges
95 all_issue_ids = set(graph.issues.keys())
96 for issue in issues:
97 for blocker_id in issue.blocked_by:
98 # Skip completed blockers (already satisfied)
99 if blocker_id in completed:
100 continue
101 # Skip blockers not in the graph
102 if blocker_id not in all_issue_ids:
103 # Only warn if the issue truly doesn't exist on disk
104 if all_known_ids is None or blocker_id not in all_known_ids:
105 logger.warning(
106 f"Issue {issue.issue_id} blocked by unknown issue {blocker_id}"
107 )
108 continue
109 # Add bidirectional edge
110 graph.blocked_by[issue.issue_id].add(blocker_id)
111 graph.blocks[blocker_id].add(issue.issue_id)
113 # Second pass: honour one-sided blocks: declarations.
114 # If A declares blocks: [B] without a matching blocked_by: [A] on B,
115 # the A→B edge was never added above; add it now (dedup guard prevents
116 # double-counting symmetric declarations).
117 for issue in issues:
118 for blocked_id in issue.blocks:
119 if blocked_id in completed:
120 continue
121 if blocked_id not in all_issue_ids:
122 if all_known_ids is None or blocked_id not in all_known_ids:
123 logger.warning(f"Issue {issue.issue_id} blocks unknown issue {blocked_id}")
124 continue
125 if issue.issue_id not in graph.blocked_by.get(blocked_id, set()):
126 graph.blocked_by[blocked_id].add(issue.issue_id)
127 graph.blocks[issue.issue_id].add(blocked_id)
129 # Third pass: populate depends_on_edges (soft prerequisite relationships).
130 # depends_on is one-directional — no reverse edge is built here.
131 for issue in issues:
132 for target_id in issue.depends_on:
133 if target_id in completed:
134 continue
135 if target_id not in all_issue_ids:
136 if all_known_ids is None or target_id not in all_known_ids:
137 logger.warning(
138 f"Issue {issue.issue_id} has depends_on unknown issue {target_id}"
139 )
140 continue
141 if issue.issue_id not in graph.depends_on_edges:
142 graph.depends_on_edges[issue.issue_id] = set()
143 graph.depends_on_edges[issue.issue_id].add(target_id)
145 return graph
147 def get_ready_issues(self, completed: set[str] | None = None) -> list[IssueInfo]:
148 """Return issues whose blockers are all completed.
150 An issue is "ready" if:
151 - It is not already completed
152 - All its blockers are either completed or not in the graph
154 Args:
155 completed: Set of completed issue IDs
157 Returns:
158 List of IssueInfo for issues with no active blockers,
159 sorted by priority (highest first) then issue_id
160 """
161 completed = completed or set()
162 ready = []
163 for issue_id, issue in self.issues.items():
164 if issue_id in completed:
165 continue
166 blockers = self.get_blocking_issues(issue_id, completed)
167 if not blockers:
168 ready.append(issue)
169 # Sort by priority then issue_id for consistent ordering
170 ready.sort(key=lambda x: (x.priority_int, x.issue_id))
171 return ready
173 def get_execution_waves(self, completed: set[str] | None = None) -> list[list[IssueInfo]]:
174 """Return issues grouped into parallel execution waves.
176 Wave 1: All issues with no blockers (or blockers already completed)
177 Wave 2: Issues whose blockers are all in wave 1
178 Wave N: Issues whose blockers are all in waves 1..N-1
180 This is similar to topological_sort but groups issues by "level"
181 rather than returning a flat list.
183 Args:
184 completed: Set of already-completed issue IDs
186 Returns:
187 List of waves, each wave is a list of issues that can run in parallel.
188 Empty list if graph is empty or all issues are completed.
190 Raises:
191 ValueError: If graph contains cycles (not a DAG)
193 Example:
194 If A blocks B and C, and B and C block D:
195 - Wave 1: [A]
196 - Wave 2: [B, C]
197 - Wave 3: [D]
198 """
199 completed = completed or set()
200 waves: list[list[IssueInfo]] = []
201 processed: set[str] = set(completed)
203 while True:
204 # Get issues ready to run (all hard blockers in processed set)
205 wave = self.get_ready_issues(completed=processed)
206 if not wave:
207 break
209 # Soft ordering: nudge depends_on targets into this wave when their
210 # hard blockers are all satisfied by processed ∪ current wave.
211 after_wave: set[str] = processed | {i.issue_id for i in wave}
212 nudged: list[IssueInfo] = []
213 for issue in wave:
214 for target_id in self.depends_on_edges.get(issue.issue_id, set()):
215 if target_id in after_wave or target_id not in self.issues:
216 continue
217 if not self.get_blocking_issues(target_id, after_wave):
218 nudged.append(self.issues[target_id])
219 after_wave.add(target_id)
220 wave.extend(nudged)
222 waves.append(wave)
223 # Mark this wave (including nudged) as processed for next iteration
224 for issue in wave:
225 processed.add(issue.issue_id)
227 # Check for cycles - if we have unprocessed issues, there's a cycle
228 remaining = set(self.issues.keys()) - processed
229 if remaining:
230 cycles = self.detect_cycles()
231 cycle_str = ", ".join(" -> ".join(cycle) for cycle in cycles)
232 raise ValueError(f"Dependency graph contains cycles: {cycle_str}")
234 return waves
236 def is_blocked(self, issue_id: str, completed: set[str] | None = None) -> bool:
237 """Check if an issue is still blocked.
239 Args:
240 issue_id: Issue ID to check
241 completed: Set of completed issue IDs
243 Returns:
244 True if issue has unresolved blockers, False otherwise
245 """
246 return bool(self.get_blocking_issues(issue_id, completed))
248 def get_blocking_issues(self, issue_id: str, completed: set[str] | None = None) -> set[str]:
249 """Return incomplete issues blocking this one.
251 Args:
252 issue_id: Issue ID to check
253 completed: Set of completed issue IDs
255 Returns:
256 Set of issue IDs still blocking this issue
257 """
258 completed = completed or set()
259 blockers = self.blocked_by.get(issue_id, set())
260 return blockers - completed
262 def get_blocked_by_issue(self, issue_id: str) -> set[str]:
263 """Return issues that this issue blocks.
265 Args:
266 issue_id: Issue ID to check
268 Returns:
269 Set of issue IDs that are blocked by this issue
270 """
271 return self.blocks.get(issue_id, set()).copy()
273 def topological_sort(self) -> list[IssueInfo]:
274 """Return issues in dependency order (Kahn's algorithm).
276 Issues with no dependencies come first, followed by issues
277 whose dependencies have been satisfied. Within each "level",
278 issues are sorted by priority then issue_id.
280 Returns:
281 List of IssueInfo in topological order
283 Raises:
284 ValueError: If graph contains cycles (not a DAG)
286 Example:
287 If A blocks B, and B blocks C, returns [A, B, C]
288 """
289 # Calculate in-degree for each node (number of blockers)
290 in_degree: dict[str, int] = {
291 issue_id: len(blockers) for issue_id, blockers in self.blocked_by.items()
292 }
294 # Start with nodes that have no blockers, sorted by priority
295 zero_degree = [
296 self.issues[issue_id] for issue_id, degree in in_degree.items() if degree == 0
297 ]
298 zero_degree.sort(key=lambda x: (x.priority_int, x.issue_id))
299 queue: deque[str] = deque(issue.issue_id for issue in zero_degree)
301 result: list[IssueInfo] = []
302 while queue:
303 issue_id = queue.popleft()
304 result.append(self.issues[issue_id])
306 # Reduce in-degree for nodes this one blocks
307 # Collect newly ready nodes, then sort before adding to queue
308 newly_ready: list[IssueInfo] = []
309 for blocked_id in self.blocks.get(issue_id, set()):
310 in_degree[blocked_id] -= 1
311 if in_degree[blocked_id] == 0:
312 newly_ready.append(self.issues[blocked_id])
314 # Sort newly ready by priority for consistent ordering
315 newly_ready.sort(key=lambda x: (x.priority_int, x.issue_id))
316 for issue in newly_ready:
317 queue.append(issue.issue_id)
319 # Check for cycles - if we didn't process all nodes, there's a cycle
320 if len(result) != len(self.issues):
321 cycles = self.detect_cycles()
322 cycle_str = ", ".join(" -> ".join(cycle) for cycle in cycles)
323 raise ValueError(f"Dependency graph contains cycles: {cycle_str}")
325 return result
327 def detect_cycles(self) -> list[list[str]]:
328 """Detect and return any dependency cycles.
330 Uses DFS with coloring to find back edges indicating cycles.
331 A cycle exists when we encounter a node that is currently being
332 visited (GRAY state) in the DFS traversal.
334 Returns:
335 List of cycles, each cycle is a list of issue IDs forming
336 a path from the cycle start back to itself.
337 Empty list if no cycles exist.
339 Example:
340 If A -> B -> C -> A (circular), returns [["A", "B", "C", "A"]]
341 """
342 WHITE, GRAY, BLACK = 0, 1, 2
343 color: dict[str, int] = dict.fromkeys(self.issues, WHITE)
344 cycles: list[list[str]] = []
345 path: list[str] = []
347 def dfs(node: str) -> None:
348 color[node] = GRAY
349 path.append(node)
351 # Traverse blockers (edges point from blocked to blocker)
352 for neighbor in self.blocked_by.get(node, set()):
353 if neighbor not in color:
354 continue
355 if color[neighbor] == GRAY:
356 # Found cycle - extract from path
357 cycle_start = path.index(neighbor)
358 cycle = path[cycle_start:] + [neighbor]
359 cycles.append(cycle)
360 elif color[neighbor] == WHITE:
361 dfs(neighbor)
363 path.pop()
364 color[node] = BLACK
366 for issue_id in self.issues:
367 if color[issue_id] == WHITE:
368 dfs(issue_id)
370 return cycles
372 def has_cycles(self) -> bool:
373 """Check if the graph contains any cycles.
375 Returns:
376 True if cycles exist, False otherwise
377 """
378 return len(self.detect_cycles()) > 0
380 def __len__(self) -> int:
381 """Return number of issues in the graph."""
382 return len(self.issues)
384 def __contains__(self, issue_id: str) -> bool:
385 """Check if an issue is in the graph."""
386 return issue_id in self.issues
389def refine_waves_for_contention(
390 waves: list[list[IssueInfo]],
391 *,
392 config: DependencyMappingConfig | None = None,
393) -> tuple[list[list[IssueInfo]], list[WaveContentionNote | None]]:
394 """Refine execution waves by splitting on file overlap.
396 For each wave with multiple issues, extracts file hints from issue
397 content and checks for pairwise overlaps. Overlapping issues are
398 split into sub-waves using greedy graph coloring so no two issues
399 in the same sub-wave modify the same files.
401 Args:
402 waves: Execution waves from get_execution_waves()
404 Returns:
405 Tuple of (refined_waves, contention_notes).
406 refined_waves: Wave list with contention-free sub-waves.
407 contention_notes: Parallel list (same length as refined_waves).
408 None for waves that weren't split, WaveContentionNote for sub-waves.
409 """
410 from little_loops.parallel.file_hints import FileHints, extract_file_hints
412 refined: list[list[IssueInfo]] = []
413 annotations: list[WaveContentionNote | None] = []
415 for orig_idx, wave in enumerate(waves):
416 if len(wave) <= 1:
417 refined.append(wave)
418 annotations.append(None)
419 continue
421 # Extract file hints for each issue in the wave
422 hints: dict[str, FileHints] = {}
423 for issue in wave:
424 content = issue.path.read_text() if issue.path.exists() else ""
425 hints[issue.issue_id] = extract_file_hints(content, issue.issue_id)
427 # Single pass — build conflict adjacency and collect contended paths together
428 conflicts: dict[str, set[str]] = {issue.issue_id: set() for issue in wave}
429 contended: set[str] = set()
430 for i, a in enumerate(wave):
431 for b in wave[i + 1 :]:
432 paths = hints[a.issue_id].get_overlapping_paths(hints[b.issue_id], config=config)
433 if paths:
434 conflicts[a.issue_id].add(b.issue_id)
435 conflicts[b.issue_id].add(a.issue_id)
436 contended |= paths
438 # Conservative serialization: issues with no file OR directory hints cannot prove
439 # they are safe to run in parallel with other hint-less issues. Add synthetic
440 # conflict edges so the greedy coloring places each such issue in its own sub-wave.
441 hintless_ids = [
442 issue.issue_id
443 for issue in wave
444 if not hints[issue.issue_id].files and not hints[issue.issue_id].directories
445 ]
446 has_unknown = len(hintless_ids) > 1
447 if has_unknown:
448 for i, a_id in enumerate(hintless_ids):
449 for b_id in hintless_ids[i + 1 :]:
450 conflicts[a_id].add(b_id)
451 conflicts[b_id].add(a_id)
453 # If no conflicts, keep wave as-is
454 if not any(conflicts.values()):
455 refined.append(wave)
456 annotations.append(None)
457 continue
459 contended_paths = sorted(contended)
461 # Greedy graph coloring — assign each issue the lowest color
462 # not used by any conflicting neighbor
463 color: dict[str, int] = {}
464 for issue in wave: # iterate in priority order (preserved from get_ready_issues)
465 used_colors = {color[c] for c in conflicts[issue.issue_id] if c in color}
466 c = 0
467 while c in used_colors:
468 c += 1
469 color[issue.issue_id] = c
471 # Group issues by color into sub-waves, preserving priority order
472 max_color = max(color.values())
473 total_sub_waves = max_color + 1
474 for c in range(total_sub_waves):
475 sub_wave = [issue for issue in wave if color[issue.issue_id] == c]
476 if sub_wave:
477 refined.append(sub_wave)
478 annotations.append(
479 WaveContentionNote(
480 contended_paths=contended_paths,
481 sub_wave_index=c,
482 total_sub_waves=total_sub_waves,
483 parent_wave_index=orig_idx,
484 has_unknown_hints=has_unknown,
485 )
486 )
488 reason_parts = []
489 if contended_paths:
490 reason_parts.append("file overlap")
491 if has_unknown:
492 reason_parts.append("unknown file hints")
493 reason = " + ".join(reason_parts) if reason_parts else "unknown"
494 logger.info(f" Wave split into {total_sub_waves} sub-waves due to {reason}")
496 return refined, annotations