Coverage for little_loops / learning_tests / extractor.py: 20%

1"""LLM-based extraction of external API dependencies from issue text (ENH-2209). 

2 

3Exposes ``extract_learning_targets()`` as an importable callable Python module 

4so downstream tools (ENH-2210 sprint pre-flight) can import it directly without 

5a shell-out. The default LLM call shells out through the active host CLI via 

6``resolve_host()`` (the same host abstraction used by 

7``session_store._call_llm_for_summary``) so extraction works with whichever 

8backend ``ll-auto`` / ``ll-sprint`` / ``ll-parallel`` is configured to use, not 

9just Anthropic. The ``llm_call`` parameter allows mock injection for unit tests. 

10 

11Follow the same pattern as ``gate.py`` (``is_record_stale``) — importable helper, 

12unit-testable with mock injection. 

13""" 

14 

15from __future__ import annotations 

16 

17import json 

18import logging 

19import os 

20import re 

21import subprocess 

22from collections.abc import Callable 

23from typing import TYPE_CHECKING 

24 

25from little_loops.host_runner import resolve_host 

26from little_loops.issue_parser import slugify 

27 

28if TYPE_CHECKING: 

29 from little_loops.issue_parser import IssueInfo 

30 

31logger = logging.getLogger(__name__) 

32 

33_EXTRACTION_PROMPT = """\ 

34Analyze the following issue text and identify all external packages, SDKs, or \ 

35third-party API surfaces that the implementation plan assumes behavior of. 

36 

37Include: 

38- Third-party Python packages (e.g. anthropic, requests, boto3, stripe) 

39- External APIs and services (e.g. Stripe webhooks, GitHub API) 

40- SDKs for external platforms or cloud services 

41- Non-obvious stdlib components whose contract is non-trivial (e.g. asyncio, multiprocessing) 

42 

43Exclude: 

44- Code internal to the project being built 

45- Standard Python builtins (str, dict, list, int, etc.) 

46- Contract-stable stdlib modules (os, sys, pathlib, json, re, datetime) 

47 

48For each identified dependency, provide its canonical short name only \ 

49(no version qualifier, no description). 

50 

51Return a JSON object as the LAST line of your response in exactly this format: 

52TARGETS_JSON:{{"targets": ["name1", "name2"], "count": N}} 

53 

54If there are no external dependencies, return: 

55TARGETS_JSON:{{"targets": [], "count": 0}} 

56 

57Issue text to analyze: 

58--- 

59{issue_text} 

60---\ 

61""" 

62 

63_TARGETS_JSON_RE = re.compile(r"TARGETS_JSON:(\{.*\})", re.MULTILINE) 

64 

65# Host-call timeout for the default extraction call (seconds), matching 

66# session_store._call_llm_for_summary. 

67_LLM_TIMEOUT_S = 60 

68 

69 

70def _default_llm_call(prompt: str) -> str: 

71 """Call the active host CLI for extraction and return the response prose. 

72 

73 Routes through ``resolve_host().build_blocking_json()`` (mirroring 

74 ``session_store._call_llm_for_summary``) so extraction respects the 

75 configured backend (``LL_HOST_CLI`` / ``orchestration.host_cli``) instead of 

76 instantiating the Anthropic SDK directly. ``model=None`` lets the host pick 

77 its own default model — a hardcoded Anthropic model id would fail against a 

78 non-Anthropic backend. 

79 

80 Fails soft: returns ``""`` on any host-call or parse failure (logged as a 

81 warning). The learning gate is a best-effort safety net, so a failed 

82 extraction must degrade to "no targets" rather than abort the whole run. 

83 """ 

84 try: 

85 inv = resolve_host().build_blocking_json(prompt=prompt, model=None) 

86 proc = subprocess.run( 

87 [inv.binary, *inv.args], 

88 env={**os.environ, **inv.env}, 

89 capture_output=True, 

90 text=True, 

91 timeout=_LLM_TIMEOUT_S, 

92 ) 

93 except subprocess.TimeoutExpired: 

94 logger.warning("_default_llm_call: host CLI timed out after %ds", _LLM_TIMEOUT_S) 

95 return "" 

96 except FileNotFoundError: 

97 logger.warning( 

98 "_default_llm_call: host CLI not found. Install the active host CLI (see LL_HOST_CLI)." 

99 ) 

100 return "" 

101 

102 if proc.returncode != 0: 

103 stderr_preview = proc.stderr.strip()[:200] if proc.stderr else "(no stderr)" 

104 logger.warning( 

105 "_default_llm_call: host CLI returned exit code %d (stderr: %s)", 

106 proc.returncode, 

107 stderr_preview, 

108 ) 

109 return "" 

110 

111 if not proc.stdout.strip(): 

112 logger.warning("_default_llm_call: host CLI returned empty stdout on exit 0") 

113 return "" 

114 

115 # Parse the JSON envelope and extract the prose 'result' field — the same 

116 # pattern as session_store._call_llm_for_summary. The prose still carries the 

117 # TARGETS_JSON:{...} line that _TARGETS_JSON_RE scans for downstream. 

118 try: 

119 stdout = proc.stdout.strip() 

120 try: 

121 envelope = json.loads(stdout) 

122 except json.JSONDecodeError: 

123 # Try JSONL: take the last non-empty line 

124 lines = [line for line in stdout.split("\n") if line.strip()] 

125 if not lines: 

126 raise 

127 envelope = json.loads(lines[-1]) 

128 

129 if envelope.get("subtype") == "error_max_structured_output_retries": 

130 logger.warning( 

131 "_default_llm_call: host CLI could not produce valid output after retries" 

132 ) 

133 return "" 

134 if envelope.get("is_error", False): 

135 err_text = str(envelope.get("result", "") or "")[:200] 

136 logger.warning("_default_llm_call: host CLI reported error: %s", err_text) 

137 return "" 

138 

139 result = envelope.get("result", "") 

140 return str(result) if result else "" 

141 except (json.JSONDecodeError, TypeError, ValueError) as e: 

142 raw_preview = proc.stdout[:300] if proc.stdout else "(empty)" 

143 logger.warning( 

144 "_default_llm_call: failed to parse host response: %s (raw: %s)", e, raw_preview 

145 ) 

146 return "" 

147 

148 

149def extract_learning_targets( 

150 issue_text: str, 

151 *, 

152 llm_call: Callable[[str], str] | None = None, 

153) -> list[str]: 

154 """Extract external API dependency names from issue text via LLM. 

155 

156 Returns a deduplicated list of target names. Issues with no external 

157 dependencies return an empty list; callers should omit the frontmatter 

158 field rather than writing an empty list. 

159 

160 Args: 

161 issue_text: Full issue file content (frontmatter + body). 

162 llm_call: Optional callable accepting a prompt string and returning 

163 response text. Defaults to a host-aware call via ``resolve_host()``. 

164 Inject a mock for unit tests. 

165 

166 Returns: 

167 Deduplicated list of target names (first-seen form preserved), 

168 e.g. ``["anthropic", "requests"]``. 

169 """ 

170 caller = llm_call if llm_call is not None else _default_llm_call 

171 prompt = _EXTRACTION_PROMPT.format(issue_text=issue_text) 

172 response = caller(prompt) 

173 

174 match = _TARGETS_JSON_RE.search(response) 

175 if not match: 

176 return [] 

177 

178 try: 

179 data = json.loads(match.group(1)) 

180 except (json.JSONDecodeError, KeyError): 

181 return [] 

182 

183 raw_targets: list[str] = data.get("targets") or [] 

184 seen: set[str] = set() 

185 result: list[str] = [] 

186 for t in raw_targets: 

187 name = t.strip() 

188 if not name: 

189 continue 

190 slug = slugify(name) 

191 if slug not in seen: 

192 seen.add(slug) 

193 result.append(name) 

194 return result 

195 

196 

197def resolve_learning_targets( 

198 issue: IssueInfo, 

199 *, 

200 llm_call: Callable[[str], str] | None = None, 

201) -> list[str]: 

202 """Return learning-test targets for an issue. 

203 

204 Returns ``issue.learning_tests_required`` when non-None (field-first). 

205 Falls back to JIT extraction from issue text via ``extract_learning_targets``. 

206 Returns [] on OSError (unreadable issue file). 

207 

208 The ``is not None`` sentinel is intentional: ``[]`` means "proven empty — 

209 no external deps" and must NOT trigger JIT extraction; ``None`` means 

210 "field not yet populated" and must trigger it. 

211 """ 

212 if issue.learning_tests_required is not None: 

213 return issue.learning_tests_required 

214 try: 

215 text = issue.path.read_text() 

216 except OSError: 

217 return [] 

218 return extract_learning_targets(text, llm_call=llm_call)