Coverage for lintro / ai / summary.py: 76%
141 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""AI summary service for generating high-level actionable insights.
3Takes all issues across all tools and produces a single concise summary
4with pattern analysis and prioritized recommendations. Uses a single
5API call for cost efficiency.
6"""
8from __future__ import annotations
10import json
11from collections import defaultdict
12from collections.abc import Sequence
13from pathlib import Path
14from typing import TYPE_CHECKING
16from loguru import logger
18from lintro.ai.fallback import complete_with_fallback
19from lintro.ai.models import AISummary
20from lintro.ai.paths import resolve_workspace_root, to_provider_path
21from lintro.ai.prompts import (
22 POST_FIX_SUMMARY_PROMPT_TEMPLATE,
23 SUMMARY_PROMPT_TEMPLATE,
24 SUMMARY_SYSTEM,
25)
26from lintro.ai.retry import (
27 DEFAULT_BACKOFF_FACTOR,
28 DEFAULT_BASE_DELAY,
29 DEFAULT_MAX_DELAY,
30 with_retry,
31)
32from lintro.ai.secrets import redact_secrets
33from lintro.ai.summary_params import SummaryGenParams
34from lintro.ai.token_budget import estimate_tokens
36if TYPE_CHECKING:
37 from lintro.ai.providers.base import AIResponse, BaseAIProvider
38 from lintro.models.core.tool_result import ToolResult
40__all__ = ["SummaryGenParams"]
43# -- Type helpers --------------------------------------------------------------
46def _ensure_str_list(value: object) -> list[str]:
47 """Coerce an AI response value to a list of strings.
49 Handles the case where the AI returns a plain string instead of
50 a list, or a list containing non-string items.
51 """
52 if isinstance(value, list):
53 non_str = [item for item in value if not isinstance(item, str)]
54 if non_str:
55 logger.warning(
56 "Dropped {} non-string items from AI response list",
57 len(non_str),
58 )
59 return [item for item in value if isinstance(item, str)]
60 if isinstance(value, str):
61 return [value]
62 return []
65def _build_issues_digest(
66 results: Sequence[ToolResult],
67 *,
68 workspace_root: Path | None = None,
69 max_tokens: int = 8000,
70) -> str:
71 """Build a compact textual digest of all issues across tools.
73 Groups by tool and error code, shows counts and sample locations.
74 Tracks token budget so the digest stays within *max_tokens*; when
75 the budget is nearly exhausted the remaining tools/codes are
76 summarised in a single truncation note.
78 Args:
79 results: Tool results containing parsed issues.
80 workspace_root: Optional root used for provider-safe path redaction.
81 max_tokens: Soft token budget for the entire digest (default 8000).
83 Returns:
84 Formatted digest string for inclusion in the prompt.
85 """
86 root = workspace_root or resolve_workspace_root()
87 lines: list[str] = []
88 used_tokens = 0
89 truncated = False
91 # Pre-compute per-tool issue lists so we can report omitted counts.
92 tool_entries: list[tuple[str, list[object]]] = []
93 for result in results:
94 if not result.issues or result.skipped:
95 continue
96 issues: list[object] = list(result.issues)
97 if issues:
98 tool_entries.append((result.name, issues))
100 omitted_issues = 0
101 omitted_tools = 0
103 for idx, (tool_name, issues) in enumerate(tool_entries):
104 # Group by code within this tool
105 by_code: dict[str, list[object]] = defaultdict(list)
106 for issue in issues:
107 code = getattr(issue, "code", None) or "unknown"
108 by_code[code].append(issue)
110 header = f"\n## {tool_name} ({len(issues)} issues)"
111 header_tokens = estimate_tokens(header)
112 if used_tokens + header_tokens > max_tokens:
113 # Budget exhausted — count this tool and all remaining.
114 omitted_issues += sum(len(iss) for _, iss in tool_entries[idx:])
115 omitted_tools += len(tool_entries) - idx
116 truncated = True
117 break
119 lines.append(header)
120 used_tokens += header_tokens
122 for code, code_issues in sorted(
123 by_code.items(),
124 key=lambda x: -len(x[1]),
125 ):
126 sample_locs = []
127 for iss in code_issues[:3]:
128 loc = to_provider_path(getattr(iss, "file", ""), root)
129 line_no = getattr(iss, "line", None)
130 if line_no:
131 loc += f":{line_no}"
132 sample_locs.append(loc)
133 more = f" (+{len(code_issues) - 3} more)" if len(code_issues) > 3 else ""
134 first_msg = redact_secrets(getattr(code_issues[0], "message", ""))
135 entry = (
136 f" [{code}] x{len(code_issues)}: "
137 f"{first_msg}"
138 f"\n e.g. {', '.join(sample_locs)}{more}"
139 )
140 entry_tokens = estimate_tokens(entry)
141 if used_tokens + entry_tokens > max_tokens:
142 # Count remaining codes in this tool + remaining tools.
143 seen_current = False
144 remaining_in_tool = 0
145 for c, ci in sorted(
146 by_code.items(),
147 key=lambda x: -len(x[1]),
148 ):
149 if c == code:
150 seen_current = True
151 if seen_current:
152 remaining_in_tool += len(ci)
153 omitted_issues += remaining_in_tool + sum(
154 len(iss) for _, iss in tool_entries[idx + 1 :]
155 )
156 omitted_tools += len(tool_entries) - idx - 1
157 truncated = True
158 break
160 lines.append(entry)
161 used_tokens += entry_tokens
163 if truncated:
164 break
166 if truncated:
167 note = f"\n(truncated — {omitted_issues} more issues"
168 if omitted_tools:
169 note += f" across {omitted_tools} tool{'s' if omitted_tools != 1 else ''}"
170 note += ")"
171 lines.append(note)
173 return "\n".join(lines)
176def _parse_summary_response(
177 content: str,
178 *,
179 input_tokens: int = 0,
180 output_tokens: int = 0,
181 cost_estimate: float = 0.0,
182) -> AISummary:
183 """Parse the AI summary response into an AISummary.
185 Falls back gracefully if JSON parsing fails.
187 Args:
188 content: Raw AI response content.
189 input_tokens: Tokens consumed for input.
190 output_tokens: Tokens generated for output.
191 cost_estimate: Estimated cost in USD.
193 Returns:
194 Parsed AISummary.
195 """
196 # Strip markdown code fences that AI models commonly wrap JSON in
197 cleaned = content.strip()
198 if cleaned.startswith("```"):
199 # Remove opening fence (with optional language tag)
200 first_newline = cleaned.find("\n")
201 if first_newline != -1:
202 cleaned = cleaned[first_newline + 1 :]
203 # Remove closing fence
204 if cleaned.rstrip().endswith("```"):
205 cleaned = cleaned.rstrip()[:-3].rstrip()
207 try:
208 data = json.loads(cleaned)
209 except json.JSONDecodeError:
210 logger.debug("Failed to parse AI summary response as JSON")
211 return AISummary(
212 overview=content[:500] if content else "Summary unavailable",
213 input_tokens=input_tokens,
214 output_tokens=output_tokens,
215 cost_estimate=cost_estimate,
216 )
218 if not isinstance(data, dict):
219 logger.debug("AI summary response is not a JSON object")
220 return AISummary(
221 overview=str(data)[:500] if data else "Summary unavailable",
222 input_tokens=input_tokens,
223 output_tokens=output_tokens,
224 cost_estimate=cost_estimate,
225 )
227 overview_raw = data.get("overview", "")
228 effort_raw = data.get("estimated_effort", "")
229 overview = overview_raw if isinstance(overview_raw, str) else str(overview_raw)
230 effort = effort_raw if isinstance(effort_raw, str) else str(effort_raw)
231 return AISummary(
232 overview=overview,
233 key_patterns=_ensure_str_list(data.get("key_patterns", [])),
234 priority_actions=_ensure_str_list(data.get("priority_actions", [])),
235 triage_suggestions=_ensure_str_list(data.get("triage_suggestions", [])),
236 estimated_effort=effort,
237 input_tokens=input_tokens,
238 output_tokens=output_tokens,
239 cost_estimate=cost_estimate,
240 )
243def _call_summary_provider(
244 prompt: str,
245 *,
246 provider: BaseAIProvider,
247 max_tokens: int,
248 timeout: float,
249 max_retries: int,
250 base_delay: float | None,
251 max_delay: float | None,
252 backoff_factor: float | None,
253 fallback_models: list[str] | None,
254) -> AISummary | None:
255 """Shared retry/call/parse helper for summary generation.
257 Args:
258 prompt: The formatted prompt to send to the provider.
259 provider: AI provider instance.
260 max_tokens: Maximum tokens for the response.
261 timeout: Request timeout in seconds per API call.
262 max_retries: Maximum retry attempts for transient API failures.
263 base_delay: Initial retry delay in seconds (None = use default).
264 max_delay: Maximum retry delay in seconds (None = use default).
265 backoff_factor: Retry backoff multiplier (None = use default).
266 fallback_models: Ordered list of fallback model identifiers.
268 Returns:
269 AISummary, or None if generation fails.
270 """
272 @with_retry(
273 max_retries=max_retries,
274 base_delay=base_delay if base_delay is not None else DEFAULT_BASE_DELAY,
275 max_delay=max_delay if max_delay is not None else DEFAULT_MAX_DELAY,
276 backoff_factor=(
277 backoff_factor if backoff_factor is not None else DEFAULT_BACKOFF_FACTOR
278 ),
279 )
280 def _call() -> AIResponse:
281 return complete_with_fallback(
282 provider,
283 prompt,
284 fallback_models=fallback_models,
285 system=SUMMARY_SYSTEM,
286 max_tokens=max_tokens,
287 timeout=timeout,
288 )
290 try:
291 response = _call()
292 return _parse_summary_response(
293 response.content,
294 input_tokens=response.input_tokens,
295 output_tokens=response.output_tokens,
296 cost_estimate=response.cost_estimate,
297 )
298 except Exception:
299 logger.debug("AI summary generation failed", exc_info=True)
300 return None
303def generate_summary(
304 results: Sequence[ToolResult],
305 provider: BaseAIProvider,
306 *,
307 max_tokens: int = 2048,
308 workspace_root: Path | None = None,
309 timeout: float = 60.0,
310 max_retries: int = 2,
311 base_delay: float | None = None,
312 max_delay: float | None = None,
313 backoff_factor: float | None = None,
314 fallback_models: list[str] | None = None,
315) -> AISummary | None:
316 """Generate a high-level AI summary of all issues.
318 Makes a single API call with a digest of all issues and returns
319 structured actionable insights.
321 Args:
322 results: Tool results containing parsed issues.
323 provider: AI provider instance.
324 max_tokens: Maximum tokens for the response.
325 workspace_root: Optional root used for provider-safe path redaction.
326 timeout: Request timeout in seconds per API call.
327 max_retries: Maximum retry attempts for transient API failures.
328 base_delay: Initial retry delay in seconds (None = use default).
329 max_delay: Maximum retry delay in seconds (None = use default).
330 backoff_factor: Retry backoff multiplier (None = use default).
331 fallback_models: Ordered list of fallback model identifiers
332 to try when the primary model fails with a retryable error.
334 Returns:
335 AISummary, or None if generation fails or there are no issues.
336 """
337 digest = _build_issues_digest(
338 results,
339 workspace_root=workspace_root,
340 max_tokens=8000,
341 )
342 if not digest.strip():
343 return None
345 total_issues = sum(
346 r.issues_count for r in results if r.issues_count and not r.skipped
347 )
348 tool_count = sum(1 for r in results if r.issues_count and not r.skipped)
350 prompt = SUMMARY_PROMPT_TEMPLATE.format(
351 total_issues=total_issues,
352 tool_count=tool_count,
353 issues_digest=digest,
354 )
356 return _call_summary_provider(
357 prompt,
358 provider=provider,
359 max_tokens=max_tokens,
360 timeout=timeout,
361 max_retries=max_retries,
362 base_delay=base_delay,
363 max_delay=max_delay,
364 backoff_factor=backoff_factor,
365 fallback_models=fallback_models,
366 )
369def generate_summary_from_params(
370 results: Sequence[ToolResult],
371 provider: BaseAIProvider,
372 params: SummaryGenParams,
373) -> AISummary | None:
374 """Generate a summary using a ``SummaryGenParams`` parameter object.
376 Thin wrapper around ``generate_summary`` that unpacks the params
377 object into keyword arguments.
379 Args:
380 results: Tool results containing parsed issues.
381 provider: AI provider instance.
382 params: Grouped generation parameters.
384 Returns:
385 AISummary, or None if generation fails or there are no issues.
386 """
387 return generate_summary(
388 results,
389 provider,
390 max_tokens=params.max_tokens,
391 workspace_root=params.workspace_root,
392 timeout=params.timeout,
393 max_retries=params.max_retries,
394 base_delay=params.base_delay,
395 max_delay=params.max_delay,
396 backoff_factor=params.backoff_factor,
397 fallback_models=params.fallback_models,
398 )
401def generate_post_fix_summary(
402 *,
403 applied: int,
404 rejected: int,
405 remaining_results: Sequence[ToolResult],
406 provider: BaseAIProvider,
407 max_tokens: int = 1024,
408 workspace_root: Path | None = None,
409 timeout: float = 60.0,
410 max_retries: int = 2,
411 base_delay: float | None = None,
412 max_delay: float | None = None,
413 backoff_factor: float | None = None,
414 fallback_models: list[str] | None = None,
415) -> AISummary | None:
416 """Generate a summary for the post-fix context.
418 Contextualizes what was fixed and what remains, providing
419 actionable next steps for remaining issues.
421 Args:
422 applied: Number of fixes applied.
423 rejected: Number of fixes rejected.
424 remaining_results: Tool results with remaining issues.
425 provider: AI provider instance.
426 max_tokens: Maximum tokens for the response.
427 workspace_root: Optional root used for provider-safe path redaction.
428 timeout: Request timeout in seconds per API call.
429 max_retries: Maximum retry attempts for transient API failures.
430 base_delay: Initial retry delay in seconds (None = use default).
431 max_delay: Maximum retry delay in seconds (None = use default).
432 backoff_factor: Retry backoff multiplier (None = use default).
433 fallback_models: Ordered list of fallback model identifiers
434 to try when the primary model fails with a retryable error.
436 Returns:
437 AISummary, or None if generation fails.
438 """
439 remaining_count = sum(
440 r.issues_count for r in remaining_results if r.issues_count and not r.skipped
441 )
443 digest = _build_issues_digest(
444 remaining_results,
445 workspace_root=workspace_root,
446 max_tokens=6000,
447 )
448 if not digest.strip() and remaining_count == 0:
449 # All issues resolved — no summary needed
450 return None
452 if digest.strip():
453 issues_digest = digest
454 else:
455 # remaining_count must be > 0 here (remaining_count == 0 returned above)
456 issues_digest = f"Remaining issues: {remaining_count} (details unavailable)"
458 prompt = POST_FIX_SUMMARY_PROMPT_TEMPLATE.format(
459 applied=applied,
460 rejected=rejected,
461 remaining=remaining_count,
462 issues_digest=issues_digest,
463 )
465 return _call_summary_provider(
466 prompt,
467 provider=provider,
468 max_tokens=max_tokens,
469 timeout=timeout,
470 max_retries=max_retries,
471 base_delay=base_delay,
472 max_delay=max_delay,
473 backoff_factor=backoff_factor,
474 fallback_models=fallback_models,
475 )
478def generate_post_fix_summary_from_params(
479 *,
480 applied: int,
481 rejected: int,
482 remaining_results: Sequence[ToolResult],
483 provider: BaseAIProvider,
484 params: SummaryGenParams,
485) -> AISummary | None:
486 """Generate a post-fix summary using a ``SummaryGenParams`` parameter object.
488 Thin wrapper around ``generate_post_fix_summary`` that unpacks the
489 params object into keyword arguments.
491 Args:
492 applied: Number of fixes applied.
493 rejected: Number of fixes rejected.
494 remaining_results: Tool results with remaining issues.
495 provider: AI provider instance.
496 params: Grouped generation parameters.
498 Returns:
499 AISummary, or None if generation fails.
500 """
501 return generate_post_fix_summary(
502 applied=applied,
503 rejected=rejected,
504 remaining_results=remaining_results,
505 provider=provider,
506 max_tokens=params.max_tokens,
507 workspace_root=params.workspace_root,
508 timeout=params.timeout,
509 max_retries=params.max_retries,
510 base_delay=params.base_delay,
511 max_delay=params.max_delay,
512 backoff_factor=params.backoff_factor,
513 fallback_models=params.fallback_models,
514 )