Coverage for lintro / ai / cache.py: 81%

1"""AI suggestion cache for deduplication across runs.""" 

2 

3from __future__ import annotations 

4 

5import dataclasses 

6import hashlib 

7import json 

8import os 

9import tempfile 

10import time 

11from pathlib import Path 

12 

13from lintro.ai.models import AIFixSuggestion 

14 

15CACHE_DIR = ".lintro-cache/ai" 

16DEFAULT_TTL = 3600 # 1 hour 

17DEFAULT_MAX_ENTRIES = 1000 

18 

19 

20def _cache_key( 

21 file_content: str, 

22 issue_code: str, 

23 issue_line: int, 

24 issue_message: str, 

25) -> str: 

26 """Compute a short SHA-256 hash key for a suggestion lookup.""" 

27 h = hashlib.sha256( 

28 f"{file_content}:{issue_code}:{issue_line}:{issue_message}".encode(), 

29 ).hexdigest()[:16] 

30 return h 

31 

32 

33def get_cached_suggestion( 

34 workspace_root: Path, 

35 file_content: str, 

36 issue_code: str, 

37 issue_line: int, 

38 issue_message: str, 

39 ttl: int = DEFAULT_TTL, 

40) -> AIFixSuggestion | None: 

41 """Return a cached suggestion if one exists and is not expired. 

42 

43 Args: 

44 workspace_root: Project root directory. 

45 file_content: Full file content (used for cache key). 

46 issue_code: Linter error code. 

47 issue_line: 1-based line number. 

48 issue_message: Linter message text. 

49 ttl: Time-to-live in seconds. 

50 

51 Returns: 

52 Cached AIFixSuggestion, or None if miss/expired. 

53 """ 

54 key = _cache_key(file_content, issue_code, issue_line, issue_message) 

55 cache_file = workspace_root / CACHE_DIR / f"{key}.json" 

56 if not cache_file.exists(): 

57 return None 

58 try: 

59 data = json.loads(cache_file.read_text()) 

60 except (json.JSONDecodeError, OSError): 

61 cache_file.unlink(missing_ok=True) 

62 return None 

63 if not isinstance(data, dict): 

64 cache_file.unlink(missing_ok=True) 

65 return None 

66 timestamp = data.get("timestamp") 

67 if not isinstance(timestamp, (int, float)): 

68 cache_file.unlink(missing_ok=True) 

69 return None 

70 if time.time() - timestamp > ttl: 

71 cache_file.unlink(missing_ok=True) 

72 return None 

73 suggestion = data.get("suggestion") 

74 if isinstance(suggestion, dict): 

75 # Touch the file to update its access/modification time for LRU tracking 

76 cache_file.touch() 

77 return AIFixSuggestion(**{ 

78 k: v 

79 for k, v in suggestion.items() 

80 if k in {f.name for f in dataclasses.fields(AIFixSuggestion)} 

81 }) 

82 return None 

83 

84 

85def _evict_lru(cache_dir: Path, max_entries: int) -> None: 

86 """Evict least recently used entries when cache exceeds *max_entries*. 

87 

88 Entries are sorted by file modification time; the oldest ones are 

89 removed until the number of entries is below *max_entries*. 

90 

91 Args: 

92 cache_dir: Directory containing cache JSON files. 

93 max_entries: Maximum number of entries to retain. 

94 """ 

95 # Build sorted list, skipping entries removed by concurrent processes. 

96 entries: list[tuple[float, Path]] = [] 

97 for p in cache_dir.glob("*.json"): 

98 try: 

99 entries.append((p.stat().st_mtime, p)) 

100 except OSError: 

101 continue 

102 entries.sort() 

103 to_remove = len(entries) - max_entries 

104 if to_remove <= 0: 

105 return 

106 for _, entry in entries[:to_remove]: 

107 entry.unlink(missing_ok=True) 

108 

109 

110def cache_suggestion( 

111 workspace_root: Path, 

112 file_content: str, 

113 issue_code: str, 

114 issue_line: int, 

115 issue_message: str, 

116 suggestion: AIFixSuggestion, 

117 max_entries: int = DEFAULT_MAX_ENTRIES, 

118) -> None: 

119 """Persist a suggestion to the on-disk cache. 

120 

121 When the cache directory exceeds *max_entries* files, least recently 

122 used entries (by file modification time) are evicted. 

123 

124 Args: 

125 workspace_root: Project root directory. 

126 file_content: Full file content (used for cache key). 

127 issue_code: Linter error code. 

128 issue_line: 1-based line number. 

129 issue_message: Linter message text. 

130 suggestion: AIFixSuggestion to cache. 

131 max_entries: Maximum cache entries before LRU eviction. 

132 """ 

133 key = _cache_key(file_content, issue_code, issue_line, issue_message) 

134 cache_dir = workspace_root / CACHE_DIR 

135 cache_dir.mkdir(parents=True, exist_ok=True) 

136 cache_file = cache_dir / f"{key}.json" 

137 suggestion_data = dataclasses.asdict(suggestion) 

138 payload = json.dumps({"timestamp": time.time(), "suggestion": suggestion_data}) 

139 fd, tmp_path = tempfile.mkstemp( 

140 dir=cache_file.parent, 

141 prefix=cache_file.name + ".", 

142 suffix=".tmp", 

143 ) 

144 try: 

145 with os.fdopen(fd, "w", encoding="utf-8") as f: 

146 f.write(payload) 

147 os.replace(tmp_path, cache_file) 

148 except BaseException: 

149 Path(tmp_path).unlink(missing_ok=True) 

150 raise 

151 _evict_lru(cache_dir, max_entries)