Coverage for little_loops / analytics / variance.py: 0%
114 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:27 -0500
1"""Non-discriminating evaluator detection from run history."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from pathlib import Path
7from typing import Any
10@dataclass
11class EvaluatorVariance:
12 """Variance analysis for a single evaluator state.
14 Attributes:
15 state: State name.
16 evaluator_type: Evaluator type (e.g. llm_structured, exit_code).
17 pass_count: Number of positive verdicts.
18 total: Total number of evaluate events for this state.
19 pass_rate: pass_count / total (0.0 if total == 0).
20 variance: Bernoulli variance p*(1-p).
21 recommendation: Human-readable recommendation if variance is low, else None.
22 ci: Wilson 95% CI (lower, upper) for pass_rate, or None if not computed.
23 """
25 state: str
26 evaluator_type: str
27 pass_count: int
28 total: int
29 pass_rate: float
30 variance: float
31 recommendation: str | None = None
32 ci: tuple[float, float] | None = None
34 def to_dict(self) -> dict[str, Any]:
35 """Convert to dictionary for JSON serialization."""
36 result: dict[str, Any] = {
37 "state": self.state,
38 "evaluator_type": self.evaluator_type,
39 "pass_count": self.pass_count,
40 "total": self.total,
41 "pass_rate": round(self.pass_rate, 4),
42 "variance": round(self.variance, 4),
43 }
44 if self.ci is not None:
45 result["ci_lower"] = round(self.ci[0], 4)
46 result["ci_upper"] = round(self.ci[1], 4)
47 if self.recommendation:
48 result["recommendation"] = self.recommendation
49 return result
52@dataclass
53class VarianceReport:
54 """Full variance report for a loop.
56 Attributes:
57 loop: Loop name.
58 total_runs: Number of runs analyzed.
59 states: Per-state variance results, sorted by variance ascending.
60 """
62 loop: str
63 total_runs: int
64 states: list[EvaluatorVariance] = field(default_factory=list)
66 def to_dict(self) -> dict[str, Any]:
67 """Convert to dictionary for JSON serialization."""
68 return {
69 "loop": self.loop,
70 "total_runs": self.total_runs,
71 "states": [s.to_dict() for s in self.states],
72 }
75def _correlate_verdicts(
76 events: list[dict[str, Any]],
77) -> dict[str, list[bool]]:
78 """Correlate evaluate events with state_enter events to get per-state verdicts.
80 Walks events chronologically, tracking current state from state_enter
81 events, then pairs each evaluate event with the tracked state.
83 Args:
84 events: List of event dicts from events.jsonl.
86 Returns:
87 Dict mapping state name to list of bool verdicts (True = positive).
88 """
89 from little_loops.fsm.persistence import _verdict_is_yes
91 state_verdicts: dict[str, list[bool]] = {}
92 current_state: str | None = None
94 for event in events:
95 if event.get("event") == "state_enter":
96 current_state = event.get("state")
97 elif event.get("event") == "evaluate" and current_state:
98 verdict_str = event.get("verdict", "")
99 state_verdicts.setdefault(current_state, []).append(_verdict_is_yes(verdict_str))
101 return state_verdicts
104def _generate_recommendation(
105 state: str,
106 evaluator_type: str,
107 pass_rate: float,
108 variance: float,
109 prompt: str | None = None,
110 target: Any = None,
111) -> str | None:
112 """Generate a recommendation for a low-variance evaluator.
114 Pattern-matches common failure modes:
115 - High pass-rate + llm_structured → "broaden judge criteria"
116 - 100% pass + output_numeric → "target may be too loose"
117 - 100% pass + exit_code → "command may not exercise the feature"
119 Args:
120 state: State name.
121 evaluator_type: Evaluator type string.
122 pass_rate: Observed pass rate.
123 variance: Bernoulli variance.
124 prompt: Evaluator prompt (for llm_structured).
125 target: Target value (for output_numeric).
127 Returns:
128 Recommendation string or None if evaluator appears discriminating.
129 """
130 if variance > 0.05:
131 return None
133 if pass_rate >= 0.95 and evaluator_type == "llm_structured":
134 prompt_preview = ""
135 if prompt:
136 truncated = prompt[:100] + "..." if len(prompt) > 100 else prompt
137 prompt_preview = f'\n ↳ judge prompt: "{truncated}"'
138 return (
139 f"Judge prompt may be too broad — most inputs pass trivially.{prompt_preview}\n"
140 f" Recommendation: tighten to require specific evidence "
141 f"(e.g., confidence_score increase, new codebase references added)."
142 )
144 if pass_rate >= 0.99 and evaluator_type == "output_numeric":
145 target_str = f" (target={target})" if target is not None else ""
146 return (
147 f"Target may be too loose for actual output values{target_str}.\n"
148 f" Recommendation: lower the target or inspect typical run outputs "
149 f"to find a more discriminating threshold."
150 )
152 if pass_rate >= 0.99 and evaluator_type == "exit_code":
153 return (
154 "Command may not exercise the feature — exits 0 regardless of intent.\n"
155 " Recommendation: replace with a command that fails on meaningful "
156 "conditions (e.g., grep for expected output, diff against baseline)."
157 )
159 return None
162def compute_evaluator_variance(
163 loop_name: str,
164 loops_dir: Path,
165 threshold: float = 0.05,
166 min_runs: int = 10,
167) -> VarianceReport | None:
168 """Compute per-state evaluator variance from run history.
170 Walks .loops/.history/*-{loop_name}/events.jsonl, correlates evaluate
171 events with state_enter events, computes Bernoulli variance p*(1-p)
172 per state, and generates recommendations for low-variance evaluators.
174 Args:
175 loop_name: Name of the loop to analyze.
176 loops_dir: Base directory containing .loops/.
177 threshold: Variance floor below which a state is flagged (default 0.05).
178 min_runs: Minimum runs required to compute meaningful variance (default 10).
180 Returns:
181 VarianceReport if history exists and min_runs is met, None otherwise.
182 """
183 from little_loops.fsm.persistence import HISTORY_DIR
185 history_root = loops_dir / HISTORY_DIR
186 if not history_root.exists():
187 return None
189 suffix = f"-{loop_name}"
190 all_verdicts: dict[str, list[bool]] = {}
191 run_count = 0
193 for run_dir in sorted(history_root.iterdir(), key=lambda d: d.name):
194 if not run_dir.is_dir() or not run_dir.name.endswith(suffix):
195 continue
196 events_file = run_dir / "events.jsonl"
197 if not events_file.exists():
198 continue
199 import json as _json
201 events: list[dict[str, Any]] = []
202 try:
203 for line in events_file.read_text(encoding="utf-8").splitlines():
204 line = line.strip()
205 if line:
206 try:
207 events.append(_json.loads(line))
208 except _json.JSONDecodeError:
209 pass
210 except OSError:
211 continue
213 run_verdicts = _correlate_verdicts(events)
214 for state, verdicts in run_verdicts.items():
215 all_verdicts.setdefault(state, []).extend(verdicts)
216 run_count += 1
218 if run_count < min_runs:
219 return None
221 # Load loop YAML to get evaluator configs
222 evaluator_configs: dict[str, dict[str, Any]] = {}
223 try:
224 from little_loops.cli.loop._helpers import load_loop
225 from little_loops.logger import Logger
227 fsm = load_loop(loop_name, loops_dir, Logger(verbose=False))
228 for name, state_cfg in fsm.states.items():
229 if state_cfg.evaluate is not None:
230 evaluator_configs[name] = {
231 "type": state_cfg.evaluate.type,
232 "prompt": state_cfg.evaluate.prompt,
233 "target": state_cfg.evaluate.target,
234 }
235 except (FileNotFoundError, ValueError):
236 pass
238 from little_loops.stats import wilson_ci
240 states: list[EvaluatorVariance] = []
241 for state_name in sorted(all_verdicts.keys()):
242 verdicts = all_verdicts[state_name]
243 total = len(verdicts)
244 if total == 0:
245 continue
246 pass_count = sum(1 for v in verdicts if v)
247 pass_rate = pass_count / total
248 variance = pass_rate * (1 - pass_rate)
249 ci = wilson_ci(pass_count, total)
251 config = evaluator_configs.get(state_name, {})
252 eval_type = config.get("type", "unknown")
253 recommendation = _generate_recommendation(
254 state_name,
255 eval_type,
256 pass_rate,
257 variance,
258 prompt=config.get("prompt"),
259 target=config.get("target"),
260 )
262 states.append(
263 EvaluatorVariance(
264 state=state_name,
265 evaluator_type=eval_type,
266 pass_count=pass_count,
267 total=total,
268 pass_rate=pass_rate,
269 variance=variance,
270 recommendation=recommendation,
271 ci=ci,
272 )
273 )
275 # Sort by variance ascending (lowest variance first — most suspicious)
276 states.sort(key=lambda s: s.variance)
278 return VarianceReport(
279 loop=loop_name,
280 total_runs=run_count,
281 states=states,
282 )