Coverage for little_loops / parallel / overlap_detector.py: 38%

1"""Overlap detection for parallel issue processing. 

2 

3Tracks active issue scopes and detects potential file modification conflicts 

4before dispatch to reduce merge conflicts. 

5""" 

6 

7from __future__ import annotations 

8 

9import logging 

10from dataclasses import dataclass, field 

11from threading import RLock 

12from typing import TYPE_CHECKING 

13 

14from little_loops.parallel.file_hints import FileHints, extract_file_hints 

15 

16if TYPE_CHECKING: 

17 from little_loops.config import DependencyMappingConfig 

18 from little_loops.issue_parser import IssueInfo 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23@dataclass 

24class OverlapResult: 

25 """Result of an overlap check. 

26 

27 Attributes: 

28 has_overlap: Whether overlap was detected 

29 overlapping_issues: Issue IDs that overlap 

30 overlapping_files: Specific files/paths that overlap 

31 """ 

32 

33 has_overlap: bool = False 

34 overlapping_issues: list[str] = field(default_factory=list) 

35 overlapping_files: set[str] = field(default_factory=set) 

36 

37 def __bool__(self) -> bool: 

38 """Allow using result directly in boolean context.""" 

39 return self.has_overlap 

40 

41 

42class OverlapDetector: 

43 """Detects overlapping file modifications between parallel issues. 

44 

45 Thread-safe tracking of which issues are currently being processed 

46 and what files they're expected to modify based on issue content analysis. 

47 

48 Usage: 

49 detector = OverlapDetector() 

50 

51 # Check before dispatch 

52 result = detector.check_overlap(new_issue) 

53 if result: 

54 # Handle overlap (serialize, warn, etc.) 

55 pass 

56 else: 

57 detector.register_issue(new_issue) 

58 

59 # After completion 

60 detector.unregister_issue(issue_id) 

61 """ 

62 

63 def __init__( 

64 self, 

65 config: DependencyMappingConfig | None = None, 

66 ) -> None: 

67 """Initialize the overlap detector. 

68 

69 Args: 

70 config: Optional dependency mapping config for custom thresholds. 

71 """ 

72 self._lock = RLock() 

73 self._active_hints: dict[str, FileHints] = {} 

74 self._config = config 

75 

76 def register_issue(self, issue: IssueInfo) -> FileHints: 

77 """Register an issue as actively being processed. 

78 

79 Args: 

80 issue: Issue being processed 

81 

82 Returns: 

83 FileHints extracted from the issue 

84 """ 

85 with self._lock: 

86 content = issue.path.read_text() if issue.path.exists() else "" 

87 hints = extract_file_hints(content, issue.issue_id) 

88 self._active_hints[issue.issue_id] = hints 

89 logger.debug( 

90 f"Registered {issue.issue_id} with hints: " 

91 f"files={hints.files}, dirs={hints.directories}, scopes={hints.scopes}" 

92 ) 

93 return hints 

94 

95 def unregister_issue(self, issue_id: str) -> None: 

96 """Unregister an issue when processing completes. 

97 

98 Args: 

99 issue_id: ID of the completed issue 

100 """ 

101 with self._lock: 

102 if issue_id in self._active_hints: 

103 del self._active_hints[issue_id] 

104 logger.debug(f"Unregistered {issue_id}") 

105 

106 def check_overlap(self, issue: IssueInfo) -> OverlapResult: 

107 """Check if an issue overlaps with any active issues. 

108 

109 Does NOT register the issue - call register_issue separately after 

110 deciding to proceed. 

111 

112 Args: 

113 issue: Issue to check 

114 

115 Returns: 

116 OverlapResult with overlap details 

117 """ 

118 with self._lock: 

119 content = issue.path.read_text() if issue.path.exists() else "" 

120 new_hints = extract_file_hints(content, issue.issue_id) 

121 

122 result = OverlapResult() 

123 

124 for active_id, active_hints in self._active_hints.items(): 

125 if new_hints.contends_with(active_hints, config=self._config): 

126 result.has_overlap = True 

127 result.overlapping_issues.append(active_id) 

128 # Find specific overlapping paths 

129 result.overlapping_files.update(new_hints.files & active_hints.files) 

130 result.overlapping_files.update( 

131 new_hints.directories & active_hints.directories 

132 ) 

133 

134 if result.has_overlap: 

135 overlap_desc = ( 

136 result.overlapping_files 

137 if result.overlapping_files 

138 else "scope/directory overlap" 

139 ) 

140 logger.info( 

141 f"{issue.issue_id} overlaps with {result.overlapping_issues}: {overlap_desc}" 

142 ) 

143 

144 return result 

145 

146 def get_active_issues(self) -> list[str]: 

147 """Get list of currently active issue IDs. 

148 

149 Returns: 

150 List of active issue IDs 

151 """ 

152 with self._lock: 

153 return list(self._active_hints.keys()) 

154 

155 def get_hints(self, issue_id: str) -> FileHints | None: 

156 """Get hints for a registered issue. 

157 

158 Args: 

159 issue_id: Issue ID to look up 

160 

161 Returns: 

162 FileHints if registered, None otherwise 

163 """ 

164 with self._lock: 

165 return self._active_hints.get(issue_id) 

166 

167 def clear(self) -> None: 

168 """Clear all tracked issues.""" 

169 with self._lock: 

170 self._active_hints.clear() 

171 logger.debug("Cleared all overlap tracking")