Coverage for lintro / ai / rerun.py: 88%

1"""Tool re-execution service for post-fix verification. 

2 

3Re-runs tools on files modified by AI fixes to get fresh remaining 

4issue counts, using the original tool execution cwd for consistency. 

5""" 

6 

7from __future__ import annotations 

8 

9import os 

10import threading 

11from contextlib import contextmanager 

12from pathlib import Path 

13from typing import TYPE_CHECKING 

14 

15from loguru import logger 

16 

17if TYPE_CHECKING: 

18 from collections.abc import Iterator 

19 

20 from lintro.models.core.tool_result import ToolResult 

21 from lintro.parsers.base_issue import BaseIssue 

22 

23_rerun_cwd_lock = threading.Lock() 

24 

25 

26@contextmanager 

27def _tool_cwd(cwd: str | None) -> Iterator[None]: 

28 """Context manager for tool execution in a specific cwd. 

29 

30 Uses a process-global lock because ``os.chdir`` is process-wide 

31 and tools call ``subprocess.run`` internally without a ``cwd`` param. 

32 

33 Known V1 limitation: ``os.chdir()`` is process-global, so other 

34 concurrent code sees the changed cwd even though the lock serializes 

35 reruns. Long-term fix: pass ``cwd`` to ``subprocess.run`` in the 

36 tool abstraction layer. 

37 

38 Args: 

39 cwd: Directory to chdir into, or None to skip. 

40 """ 

41 if not cwd: 

42 yield 

43 return 

44 

45 with _rerun_cwd_lock: 

46 original_cwd = Path.cwd() 

47 os.chdir(cwd) 

48 try: 

49 yield 

50 finally: 

51 os.chdir(original_cwd) 

52 

53 

54def paths_for_context( 

55 *, 

56 file_paths: list[str], 

57 cwd: str | None, 

58) -> list[str]: 

59 """Prefer paths relative to tool cwd when possible. 

60 

61 Args: 

62 file_paths: Absolute file paths to relativize. 

63 cwd: Tool's original working directory. 

64 

65 Returns: 

66 List of paths, made relative to cwd where possible. 

67 """ 

68 if not cwd: 

69 return file_paths 

70 

71 try: 

72 cwd_path = Path(cwd).resolve() 

73 except OSError: 

74 return file_paths 

75 

76 contextual_paths: list[str] = [] 

77 for file_path in file_paths: 

78 try: 

79 resolved = Path(file_path).resolve() 

80 except OSError: 

81 contextual_paths.append(file_path) 

82 continue 

83 try: 

84 contextual_paths.append(str(resolved.relative_to(cwd_path))) 

85 except ValueError: 

86 contextual_paths.append(str(resolved)) 

87 return contextual_paths 

88 

89 

90def rerun_tools( 

91 by_tool: dict[str, tuple[ToolResult, list[BaseIssue]]], 

92) -> list[ToolResult] | None: 

93 """Re-run tools on analyzed files to get fresh remaining issue counts. 

94 

95 Reuses the original tool execution cwd for path/config consistency. 

96 

97 Args: 

98 by_tool: Dict mapping tool name to (result, issues) pairs. 

99 

100 Returns: 

101 List of fresh tool results from re-running checks. 

102 """ 

103 try: 

104 from lintro.tools import tool_manager 

105 except ImportError: 

106 return None 

107 

108 rerun_results: list[ToolResult] = [] 

109 for tool_name, (result, issues) in by_tool.items(): 

110 file_paths = sorted({issue.file for issue in issues if issue.file}) 

111 if not file_paths: 

112 continue 

113 

114 rerun_paths = paths_for_context(file_paths=file_paths, cwd=result.cwd) 

115 

116 try: 

117 tool = tool_manager.get_tool(tool_name) 

118 with _tool_cwd(result.cwd): 

119 rerun_results.append(tool.check(rerun_paths, {})) 

120 except (KeyError, ImportError): 

121 logger.debug( 

122 f"AI post-fix rerun skipped for {tool_name}: tool not available", 

123 ) 

124 continue 

125 except Exception: 

126 logger.warning( 

127 f"AI post-fix rerun failed for {tool_name}", 

128 exc_info=True, 

129 ) 

130 continue 

131 return rerun_results 

132 

133 

134def apply_rerun_results( 

135 *, 

136 by_tool: dict[str, tuple[ToolResult, list[BaseIssue]]], 

137 rerun_results: list[ToolResult], 

138) -> None: 

139 """Apply fresh rerun issue counts back to original FIX results. 

140 

141 Args: 

142 by_tool: Dict mapping tool name to (result, issues) pairs. 

143 rerun_results: Fresh results from re-running tools. 

144 """ 

145 rerun_by_name = {result.name: result for result in rerun_results} 

146 

147 for tool_name, (result, _issues) in by_tool.items(): 

148 rerun = rerun_by_name.get(tool_name) 

149 if rerun is None: 

150 continue 

151 

152 refreshed_issues = list(rerun.issues) if rerun.issues is not None else [] 

153 # Preserve native fix counters — only update remaining issues 

154 # and issue list. The initial/fixed counts reflect what the native 

155 # tool originally reported and should not be zeroed. 

156 result.issues = refreshed_issues 

157 result.issues_count = len(refreshed_issues) 

158 result.remaining_issues_count = len(refreshed_issues) 

159 result.success = rerun.success 

160 if rerun.output is not None: 

161 result.output = rerun.output