Coverage for little_loops / issue_history / evolution.py: 0%
152 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""Evolution trigger detectors for analyze-history (ENH-1911).
3Queries history.db to surface recurring user corrections and skill bypasses
4as quantified signals for harness self-improvement.
5"""
7from __future__ import annotations
9import hashlib
10import logging
11import re
12import sqlite3
13from pathlib import Path
14from typing import Any
16from little_loops.config.features import EvolutionConfig
17from little_loops.history_reader import _stale_cutoff
18from little_loops.issue_history.models import (
19 RecurringFeedback,
20 RecurringFeedbackAnalysis,
21 SkillBypass,
22 SkillBypassAnalysis,
23)
25logger = logging.getLogger(__name__)
27_STALE_DAYS = 90 # Look back 90 days for evolution signals
30def _open_db(db_path: Path) -> sqlite3.Connection | None:
31 """Open *db_path* for read-only querying without running schema migrations.
33 Uses a direct URI connection so the file is opened as-is. This avoids the
34 ``ensure_db`` migration path inside ``_connect_readonly``, which fails when
35 the database was created by the test harness (tables already exist).
36 Returns ``None`` when the file does not exist.
37 """
38 if not db_path.exists():
39 return None
40 try:
41 conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
42 conn.row_factory = sqlite3.Row
43 conn.execute("PRAGMA query_only = ON")
44 return conn
45 except sqlite3.Error:
46 logger.warning("evolution: could not open %s read-only", db_path, exc_info=True)
47 return None
50_MIN_BYPASS_KEYWORDS = 2 # Require >= 2 keyword tokens to reduce false positives
53def _fingerprint(content: str) -> str:
54 """Return a stable 16-char hex fingerprint for a correction content string."""
55 return hashlib.sha256(content[:512].encode()).hexdigest()[:16]
58def _load_memory_feedback(project_root: Path) -> dict[str, str]:
59 """Load curated feedback topics and content from memory/feedback_* files."""
60 result: dict[str, str] = {}
61 memory_dir = project_root / "memory"
62 if not memory_dir.is_dir():
63 return result
64 for f in sorted(memory_dir.glob("feedback_*.md")):
65 try:
66 content = f.read_text(encoding="utf-8")
67 # Strip frontmatter
68 if content.startswith("---"):
69 end = content.find("---", 3)
70 body = content[end + 3 :].strip() if end != -1 else content.strip()
71 else:
72 body = content.strip()
73 result[f.stem] = body[:500]
74 except OSError:
75 pass
76 return result
79def _get_session_ids_for_content(conn: sqlite3.Connection, content: str, cutoff: str) -> list[str]:
80 """Return distinct session IDs containing a correction matching content."""
81 try:
82 rows = conn.execute(
83 "SELECT DISTINCT session_id FROM user_corrections "
84 "WHERE content = ? AND ts >= ? LIMIT 10",
85 (content, cutoff),
86 ).fetchall()
87 return [row["session_id"] for row in rows if row["session_id"]]
88 except sqlite3.Error:
89 return []
92def detect_recurring_feedback(
93 db_path: Path,
94 config: EvolutionConfig,
95 project_root: Path | None = None,
96) -> RecurringFeedbackAnalysis:
97 """Detect user corrections that have recurred >= config.feedback_min_recurrence times.
99 Queries user_corrections grouped by content with a HAVING threshold filter,
100 enriches each result with session IDs, and cross-references memory/feedback_*
101 files for candidate_rule seeds.
102 """
103 conn = _open_db(db_path)
104 if conn is None:
105 return RecurringFeedbackAnalysis()
107 try:
108 cutoff = _stale_cutoff(_STALE_DAYS)
109 threshold = config.feedback_min_recurrence
111 try:
112 rows = conn.execute(
113 "SELECT content, COUNT(*) AS seen_count "
114 "FROM user_corrections "
115 "WHERE ts >= ? "
116 "GROUP BY content "
117 "HAVING seen_count >= ? "
118 "ORDER BY seen_count DESC, MAX(ts) DESC "
119 "LIMIT 50",
120 (cutoff, threshold),
121 ).fetchall()
122 except sqlite3.Error:
123 logger.warning("evolution: recurring feedback query failed", exc_info=True)
124 return RecurringFeedbackAnalysis()
126 memory_feedback = _load_memory_feedback(project_root) if project_root else {}
128 # Load retirement fingerprints — fall back gracefully for old DBs missing the table.
129 try:
130 retired_rows = conn.execute(
131 "SELECT topic_fingerprint, rule_id FROM correction_retirements"
132 ).fetchall()
133 retirements: dict[str, str] = {
134 r["topic_fingerprint"]: (r["rule_id"] or "") for r in retired_rows
135 }
136 except sqlite3.OperationalError:
137 retirements = {}
139 feedbacks: list[RecurringFeedback] = []
140 retired_count = 0
141 for row in rows:
142 content = row["content"] or ""
143 count = row["seen_count"]
144 fingerprint = _fingerprint(content)
146 # Exclude clusters that have been addressed and retired.
147 if fingerprint in retirements:
148 retired_count += 1
149 continue
151 session_ids = _get_session_ids_for_content(conn, content, cutoff)
153 # Match memory feedback files by shared keywords as candidate_rule seed
154 candidate_rule = ""
155 content_words = set(re.findall(r"[a-z]{3,}", content.lower()))
156 for _mem_topic, mem_body in memory_feedback.items():
157 topic_words = set(re.findall(r"[a-z]{3,}", _mem_topic.lower()))
158 if topic_words & content_words:
159 candidate_rule = mem_body[:200]
160 break
162 excerpt = content[:120] + "..." if len(content) > 120 else content
163 feedbacks.append(
164 RecurringFeedback(
165 topic=excerpt,
166 occurrence_count=count,
167 example_sessions=session_ids[:5],
168 example_content=[content[:200]],
169 candidate_rule=candidate_rule,
170 topic_fingerprint=fingerprint,
171 )
172 )
174 rule_candidates = [f.candidate_rule for f in feedbacks if f.candidate_rule][:10]
175 return RecurringFeedbackAnalysis(
176 feedbacks=feedbacks,
177 total_recurring_corrections=sum(f.occurrence_count for f in feedbacks),
178 threshold_used=threshold,
179 rule_candidates=rule_candidates,
180 retired_count=retired_count,
181 )
182 finally:
183 conn.close()
186def _load_skill_keywords(project_root: Path) -> dict[str, set[str]]:
187 """Load keyword sets for all registered skills."""
188 from little_loops.cli.verify_triggers import _extract_keywords, _load_skill_descriptions
190 skills_dir = project_root / "skills"
191 descriptions = _load_skill_descriptions(skills_dir)
192 return {name: _extract_keywords(desc) for name, (desc, _path) in descriptions.items()}
195def _tokenize_content(text: str) -> set[str]:
196 """Tokenize message content for keyword matching."""
197 from little_loops.cli.verify_triggers import _tokenize
199 return _tokenize(text[:512])
202def detect_skill_bypass(
203 db_path: Path,
204 config: EvolutionConfig,
205 project_root: Path | None = None,
206) -> SkillBypassAnalysis:
207 """Detect sessions where user manually performed work a skill covers.
209 Compares message_events content against skill keyword sets. A bypass is counted
210 when >= _MIN_BYPASS_KEYWORDS keyword tokens match AND no skill_events row for
211 that skill exists in the same session. Conservative threshold reduces false positives.
212 """
213 if project_root is None:
214 return SkillBypassAnalysis()
216 conn = _open_db(db_path)
217 if conn is None:
218 return SkillBypassAnalysis()
220 try:
221 cutoff = _stale_cutoff(_STALE_DAYS)
222 threshold = config.bypass_min_count
224 skill_keywords = _load_skill_keywords(project_root)
225 if not skill_keywords:
226 return SkillBypassAnalysis()
228 try:
229 messages = conn.execute(
230 "SELECT session_id, content FROM message_events WHERE ts >= ? LIMIT 5000",
231 (cutoff,),
232 ).fetchall()
234 skill_rows = conn.execute(
235 "SELECT skill_name, session_id FROM skill_events WHERE ts >= ?",
236 (cutoff,),
237 ).fetchall()
238 except sqlite3.Error:
239 logger.warning("evolution: skill bypass query failed", exc_info=True)
240 return SkillBypassAnalysis()
242 # Build invocation map: skill_name -> set of session_ids where skill WAS invoked
243 skill_invocations: dict[str, set[str]] = {}
244 for row in skill_rows:
245 skill_name = row["skill_name"]
246 session_id = row["session_id"]
247 if skill_name and session_id:
248 skill_invocations.setdefault(skill_name, set()).add(session_id)
250 # Per-skill bypass accumulators
251 bypass_data: dict[str, dict[str, Any]] = {
252 name: {"count": 0, "sessions": [], "evidence": []} for name in skill_keywords
253 }
254 # Dedupe: only count each (skill, session) pair once
255 seen_pairs: dict[str, set[str]] = {name: set() for name in skill_keywords}
257 for msg in messages:
258 session_id = msg["session_id"]
259 content = msg["content"] or ""
260 if not session_id or not content:
261 continue
263 tokens = _tokenize_content(content)
264 if not tokens:
265 continue
267 for skill_name, keywords in skill_keywords.items():
268 if session_id in seen_pairs[skill_name]:
269 continue
270 # Require >= _MIN_BYPASS_KEYWORDS matching tokens (conservative)
271 if len(tokens & keywords) < _MIN_BYPASS_KEYWORDS:
272 continue
273 # Check if skill was invoked in this session
274 if session_id in skill_invocations.get(skill_name, set()):
275 continue
276 # Bypass detected
277 seen_pairs[skill_name].add(session_id)
278 data = bypass_data[skill_name]
279 data["count"] += 1
280 if len(data["sessions"]) < 5:
281 data["sessions"].append(session_id)
282 if len(data["evidence"]) < 3:
283 data["evidence"].append(content[:150])
285 bypasses: list[SkillBypass] = []
286 for skill_name, data in bypass_data.items():
287 if data["count"] >= threshold:
288 bypasses.append(
289 SkillBypass(
290 skill_name=skill_name,
291 bypass_count=data["count"],
292 example_sessions=data["sessions"],
293 evidence=data["evidence"],
294 suggested_improvement=(
295 f"Review trigger keywords for '{skill_name}' — "
296 f"users are doing this manually {data['count']}x"
297 ),
298 )
299 )
301 bypasses.sort(key=lambda b: -b.bypass_count)
302 suggestions = [
303 f"Sharpen trigger for '{b.skill_name}' (bypassed {b.bypass_count}x)" for b in bypasses
304 ][:10]
306 return SkillBypassAnalysis(
307 bypasses=bypasses,
308 total_bypassed_invocations=sum(b.bypass_count for b in bypasses),
309 threshold_used=threshold,
310 improvement_suggestions=suggestions,
311 )
312 finally:
313 conn.close()