Coverage for lintro / ai / telemetry.py: 100%

1"""AI session telemetry collector. 

2 

3Accumulates per-call metrics (tokens, cost, latency) and fix outcomes 

4across the entire AI session so they can be attached to result metadata 

5at the end of the pipeline. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass 

11from typing import Any 

12 

13 

14@dataclass 

15class AITelemetry: 

16 """Accumulator for AI session metrics. 

17 

18 Attributes: 

19 total_api_calls: Number of AI provider API calls made. 

20 total_input_tokens: Sum of input tokens across all calls. 

21 total_output_tokens: Sum of output tokens across all calls. 

22 total_cost_usd: Cumulative estimated cost in USD. 

23 total_latency_ms: Cumulative latency in milliseconds. 

24 successful_fixes: Number of fixes that were applied successfully. 

25 failed_fixes: Number of fixes that failed to apply. 

26 """ 

27 

28 total_api_calls: int = 0 

29 total_input_tokens: int = 0 

30 total_output_tokens: int = 0 

31 total_cost_usd: float = 0.0 

32 total_latency_ms: float = 0.0 

33 successful_fixes: int = 0 

34 failed_fixes: int = 0 

35 

36 def record_call( 

37 self, 

38 *, 

39 input_tokens: int, 

40 output_tokens: int, 

41 cost: float, 

42 latency_ms: float, 

43 ) -> None: 

44 """Record a single AI provider API call. 

45 

46 Args: 

47 input_tokens: Input tokens consumed by the call. 

48 output_tokens: Output tokens generated by the call. 

49 cost: Estimated cost in USD for the call. 

50 latency_ms: Latency in milliseconds for the call. 

51 """ 

52 self.total_api_calls += 1 

53 self.total_input_tokens += input_tokens 

54 self.total_output_tokens += output_tokens 

55 self.total_cost_usd += cost 

56 self.total_latency_ms += latency_ms 

57 

58 def to_dict(self) -> dict[str, Any]: 

59 """Serialize telemetry to a JSON-compatible dictionary. 

60 

61 Returns: 

62 Dictionary with all telemetry fields, cost rounded to 

63 6 decimal places and latency rounded to 1 decimal place. 

64 """ 

65 return { 

66 "api_calls": self.total_api_calls, 

67 "input_tokens": self.total_input_tokens, 

68 "output_tokens": self.total_output_tokens, 

69 "cost_usd": round(self.total_cost_usd, 6), 

70 "latency_ms": round(self.total_latency_ms, 1), 

71 "successful_fixes": self.successful_fixes, 

72 "failed_fixes": self.failed_fixes, 

73 }