Coverage for little_loops / cli / history_context.py: 10%
211 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""ll-history-context: render a Historical Context block for an issue.
3Queries .ll/history.db for user corrections and FTS5 matches relevant to an
4issue ID and prints a ready-to-inject ``## Historical Context`` markdown block.
5Returns empty output when the DB is missing, has no matches, or all rows are
6stale.
8Pass ``--project`` instead of an issue ID to print the project-wide context
9digest that the session-start hook would inject (inspection / config-tuning
10dry-run, ENH-1907).
12Usage:
13 ll-history-context <issue_id> [--file <path>] [--db <path>]
14 ll-history-context --project [--db <path>]
15"""
17from __future__ import annotations
19import argparse
20import hashlib
21import logging
22import re
23import sys
24from datetime import UTC, datetime, timedelta
25from pathlib import Path
27import yaml
29from little_loops.cli.output import configure_output, use_color_enabled
30from little_loops.config.core import resolve_config_path
31from little_loops.history_reader import (
32 STALE_DAYS_DEFAULT,
33 SearchResult,
34 UserCorrection,
35 find_user_corrections,
36 issue_effort,
37 project_digest,
38 recent_file_events,
39 render_project_context,
40 search,
41)
42from little_loops.logger import Logger
43from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context, connect
45logger = logging.getLogger(__name__)
47_MAX_ROWS = 5
50def _content_hash(content: str) -> str:
51 return hashlib.sha256(content.encode()).hexdigest()[:16]
54def _render_learning_test_section(
55 targets: list[str],
56 *,
57 stale_after_days: int = 30,
58 base_dir: Path | None = None,
59) -> str | None:
60 """Build a ## Learning Test Evidence markdown table, or None if no records exist."""
61 from little_loops.learning_tests import check_learning_test
62 from little_loops.learning_tests.gate import is_record_stale
64 rows: list[str] = []
65 for target in targets:
66 record = check_learning_test(target, base_dir=base_dir)
67 if record is None:
68 continue
69 effective_status = "stale" if is_record_stale(record, stale_after_days) else record.status
70 passes = sum(1 for a in record.assertions if a.result == "pass")
71 fails = sum(1 for a in record.assertions if a.result == "fail")
72 untested = sum(1 for a in record.assertions if a.result == "untested")
73 rows.append(
74 f"| {target} | {effective_status} | {record.date} | {passes}/{fails}/{untested} |"
75 )
77 if not rows:
78 return None
80 lines = [
81 "## Learning Test Evidence",
82 "",
83 "| Target | Status | Date | Pass/Fail/Untested |",
84 "|--------|--------|------|--------------------|",
85 *rows,
86 ]
87 return "\n".join(lines)
90def _get_issue_lt_targets(issue_id: str, cfg: object) -> list[str]:
91 """Read learning_tests_required from the issue file frontmatter."""
92 from little_loops.cli.issues.show import _resolve_issue_id
94 issue_path = _resolve_issue_id(cfg, issue_id) # type: ignore[arg-type]
95 if issue_path is None:
96 return []
97 try:
98 content = issue_path.read_text(encoding="utf-8")
99 except OSError:
100 return []
101 fm_match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
102 if not fm_match:
103 return []
104 try:
105 fm = yaml.safe_load(fm_match.group(1)) or {}
106 except yaml.YAMLError:
107 return []
108 raw = fm.get("learning_tests_required")
109 if not raw:
110 return []
111 if isinstance(raw, list):
112 return [str(t) for t in raw]
113 if isinstance(raw, str):
114 return [raw]
115 return []
118def _build_parser() -> argparse.ArgumentParser:
119 parser = argparse.ArgumentParser(
120 prog="ll-history-context",
121 description="Render a ## Historical Context block for an issue from .ll/history.db",
122 formatter_class=argparse.RawDescriptionHelpFormatter,
123 epilog="""
124Examples:
125 %(prog)s ENH-1708 # Corrections matching the issue ID
126 %(prog)s ENH-1708 --file src/foo.py # Also include recent file events
127 %(prog)s ENH-1708 --db custom/history.db # Use a non-default database
128 %(prog)s --project # Print project-wide context digest
129""",
130 )
131 parser.add_argument(
132 "issue_id",
133 metavar="ISSUE_ID",
134 nargs="?",
135 default=None,
136 help="Issue ID to query (e.g. ENH-1708). Mutually exclusive with --project.",
137 )
138 parser.add_argument(
139 "--project",
140 action="store_true",
141 default=False,
142 help="Print the project-wide context digest (dry-run of session-start injection).",
143 )
144 parser.add_argument(
145 "--file",
146 metavar="PATH",
147 default=None,
148 help="Also include recent file events for this path",
149 )
150 parser.add_argument(
151 "--db",
152 type=Path,
153 default=DEFAULT_DB_PATH,
154 metavar="PATH",
155 help="Path to the session database (default: .ll/history.db)",
156 )
157 parser.add_argument(
158 "--effort",
159 action="store_true",
160 default=False,
161 help="Include effort/velocity context (session count and cycle time)",
162 )
163 parser.add_argument(
164 "--for-skill",
165 type=str,
166 default=None,
167 metavar="NAME",
168 help="Exit 0 with no output if NAME is not in history.planning_skills.",
169 )
170 return parser
173def main_history_context() -> int:
174 """Entry point for ll-history-context command.
176 Returns:
177 0 on success (including empty output when no matches or DB absent), 1 on argument error.
178 """
179 with cli_event_context(DEFAULT_DB_PATH, "ll-history-context", sys.argv[1:]):
180 configure_output()
181 logger = Logger(use_color=use_color_enabled()) # noqa: F841
183 parser = _build_parser()
184 args = parser.parse_args()
186 # Mutual-exclusion guard: require exactly one of issue_id or --project.
187 if args.project and args.issue_id:
188 parser.error("--project and ISSUE_ID are mutually exclusive")
189 if not args.project and not args.issue_id:
190 parser.error("one of ISSUE_ID or --project is required")
192 # --project: print the project-wide digest dry-run and exit.
193 if args.project:
194 from little_loops.config.features import HistoryConfig
196 cwd = Path.cwd()
197 config_path = resolve_config_path(cwd)
198 if config_path is not None:
199 import json
201 try:
202 merged = json.loads(config_path.read_text(encoding="utf-8"))
203 except (OSError, json.JSONDecodeError):
204 merged = {}
205 else:
206 merged = {}
208 _hist = HistoryConfig.from_dict(merged.get("history", {}))
209 _sd = _hist.session_digest
210 _sections = _sd.sections if _sd.sections else None
211 digest = project_digest(args.db, days=_sd.days, sections=_sections)
212 block = render_project_context(digest, char_cap=_sd.char_cap, days=_sd.days)
213 if block:
214 print(block)
215 return 0
217 # --for-skill guard: exit 0 with no output if skill is not in history.planning_skills.
218 if args.for_skill is not None:
219 from little_loops.config import BRConfig
221 cfg = BRConfig(Path.cwd())
222 if args.for_skill not in cfg.history.planning_skills:
223 return 0
225 # --effort: print effort/velocity context and exit.
226 if args.effort:
227 from little_loops.config import BRConfig
229 cfg = BRConfig(Path.cwd())
230 fields = cfg.history.effort_fields
231 effort = issue_effort(args.issue_id, db=args.db)
232 if effort is None:
233 return 0
234 valid_fields = {"session_count", "cycle_time_days"}
235 print("## Effort Context")
236 print()
237 for f in fields:
238 if f not in valid_fields:
239 logger.warning(f"history_context: unknown effort field {f!r} — skipping")
240 continue
241 val = effort.get(f)
242 if val is None:
243 print(f"- {f}: n/a")
244 else:
245 print(f"- {f}: {val}")
246 return 0
248 cutoff = (datetime.now(UTC) - timedelta(days=STALE_DAYS_DEFAULT)).strftime(
249 "%Y-%m-%dT%H:%M:%SZ"
250 )
252 corrections: list[UserCorrection] = find_user_corrections(
253 topic=args.issue_id, limit=20, db=args.db
254 )
256 search_results: list[SearchResult] = search(
257 query=args.issue_id, kind="correction", limit=20, db=args.db
258 )
259 fresh_search = [r for r in search_results if r.ts >= cutoff]
261 file_contents: list[str] = []
262 if args.file:
263 for fe in recent_file_events(path=args.file, limit=5, db=args.db):
264 file_contents.append(f"file:{fe.path}:{fe.op}")
266 seen: set[str] = set()
267 rows: list[str] = []
269 for uc in corrections:
270 h = _content_hash(uc.content)
271 if h not in seen:
272 seen.add(h)
273 rows.append(uc.content)
275 for sr in fresh_search:
276 h = _content_hash(sr.content)
277 if h not in seen:
278 seen.add(h)
279 rows.append(sr.content)
281 for fc in file_contents:
282 h = _content_hash(fc)
283 if h not in seen:
284 seen.add(h)
285 rows.append(fc)
287 rows = rows[:_MAX_ROWS]
289 if not rows:
290 # Fallback: query issue_snapshots when no corrections/search rows found.
291 try:
292 conn = connect(args.db)
293 try:
294 snap = conn.execute(
295 "SELECT title, body FROM issue_snapshots"
296 " WHERE issue_id = ? ORDER BY ts DESC LIMIT 1",
297 (args.issue_id,),
298 ).fetchone()
299 finally:
300 conn.close()
301 if snap is not None:
302 body_text = (snap["body"] or "").strip()
303 if body_text:
304 rows.append(body_text[:512])
305 except Exception:
306 pass
308 # Build Learning Test Evidence section if feature is enabled.
309 lt_section: str | None = None
310 try:
311 from little_loops.config import BRConfig
313 cfg = BRConfig(Path.cwd())
314 if cfg.learning_tests.enabled:
315 targets = _get_issue_lt_targets(args.issue_id, cfg)
316 if targets:
317 lt_dir = Path.cwd() / ".ll" / "learning-tests"
318 lt_section = _render_learning_test_section(
319 targets,
320 stale_after_days=cfg.learning_tests.stale_after_days,
321 base_dir=lt_dir,
322 )
323 except Exception:
324 pass
326 # Build Prior Work (condensed) section if compaction is enabled (ENH-2231).
327 prior_work_section: str | None = None
328 _PRIOR_WORK_SECTION_BUDGET = 1000
329 _PRIOR_WORK_NODE_CHAR_CAP = 400
330 try:
331 from little_loops.config import BRConfig
332 from little_loops.history_reader import condensed_nodes_for_issue
334 _cfg = BRConfig(Path.cwd())
335 if _cfg.history.compaction.enabled:
336 _nodes = condensed_nodes_for_issue(
337 args.issue_id,
338 node_char_cap=_PRIOR_WORK_NODE_CHAR_CAP,
339 db=args.db,
340 )
341 if _nodes:
342 section_lines: list[str] = ["## Prior Work (condensed)", ""]
343 budget = _PRIOR_WORK_SECTION_BUDGET
344 for node in _nodes:
345 text = (node.content or "").strip()
346 provenance = (
347 f"*(session: {node.session_id or 'unknown'},"
348 f" ts_end: {node.ts_end or 'unknown'})*"
349 )
350 entry = f"{text}\n\n{provenance}"
351 if budget <= 0:
352 break
353 section_lines.append(entry)
354 section_lines.append("")
355 budget -= len(entry)
356 if len(section_lines) > 2:
357 prior_work_section = "\n".join(section_lines).rstrip()
358 except Exception:
359 pass
361 if not rows and lt_section is None and prior_work_section is None:
362 return 0
364 if rows:
365 print("## Historical Context")
366 print()
367 for row in rows:
368 print(f"- {row}")
370 if lt_section is not None:
371 if rows:
372 print()
373 print(lt_section)
375 if prior_work_section is not None:
376 if rows or lt_section is not None:
377 print()
378 print(prior_work_section)
380 return 0