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