Coverage for little_loops / history_reader.py: 24%
444 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -0500
1"""Typed read-only query module for ``.ll/history.db`` (ENH-1752).
3Provides the common queries that ll skills and agents need to consume the
4session database without importing ad-hoc SQL into every caller. All
5functions degrade gracefully: missing/empty/corrupt databases return empty
6lists, never raise.
8Public API:
9 UserCorrection: dataclass for user correction rows
10 FileEvent: dataclass for file event rows
11 SearchResult: dataclass for FTS5 search results
12 IssueEvent: dataclass for issue event rows
13 SessionRef: dataclass for issue_sessions view rows (ENH-1711)
14 SummaryNode: dataclass for summary_nodes rows (FEAT-1712)
15 GrepResult: dataclass for ll_grep results with summary context (FEAT-1712)
16 SectionProvider: config-addressable digest section (ENH-1907)
17 ProjectDigest: aggregated project-context snapshot (ENH-1907)
18 SECTION_PROVIDERS: registry of v1 section providers (ENH-1907)
19 find_user_corrections(topic, ...) -> list[UserCorrection]
20 recent_file_events(path, ...) -> list[FileEvent]
21 search(query, ...) -> list[SearchResult]
22 related_issue_events(issue_id, ...) -> list[IssueEvent]
23 sessions_for_issue(issue_id, ...) -> list[SessionRef]
24 issue_effort(issue_id, ...) -> dict | None
25 recent_issue_velocity(limit, ...) -> list[dict]
26 lookup_session_metadata(session_id, ...) -> dict
27 conversation_turns(db_path, ...) -> list[list[tuple[str, str]]]
28 ll_grep(pattern, ...) -> list[GrepResult]
29 ll_expand(summary_id, ...) -> list[dict]
30 ll_describe(node_id, ...) -> SummaryNode | None
31 condensed_nodes_for_issue(issue_id, ...) -> list[SummaryNode]
32 project_digest(db_path, ...) -> ProjectDigest
33 render_project_context(digest, ...) -> str
34"""
36from __future__ import annotations
38import logging
39import re
40import sqlite3
41from collections.abc import Callable
42from dataclasses import dataclass
43from datetime import UTC, datetime, timedelta
44from pathlib import Path
45from typing import Any
47from little_loops.session_store import DEFAULT_DB_PATH, ensure_db
49logger = logging.getLogger(__name__)
51STALE_DAYS_DEFAULT = 30
54# ---------------------------------------------------------------------------
55# Dataclasses
56# ---------------------------------------------------------------------------
59@dataclass
60class UserCorrection:
61 ts: str
62 session_id: str | None
63 content: str
64 source: str | None
67@dataclass
68class FileEvent:
69 ts: str
70 session_id: str | None
71 path: str | None
72 op: str | None
73 issue_id: str | None
74 git_sha: str | None
77@dataclass
78class SearchResult:
79 content: str
80 kind: str
81 ref: str
82 anchor: str
83 ts: str
84 score: float
87@dataclass
88class IssueEvent:
89 ts: str
90 issue_id: str | None
91 transition: str | None
92 discovered_by: str | None
93 issue_type: str | None
94 priority: str | None
97@dataclass
98class SessionRef:
99 """A session that co-occurred with an issue's active period (ENH-1711)."""
101 issue_id: str | None
102 session_id: str | None
103 jsonl_path: str | None
104 first_message_ts: str | None
105 last_message_ts: str | None
108@dataclass
109class SummaryNode:
110 """A summary_nodes row from the LCM-style compaction DAG (FEAT-1712)."""
112 id: int
113 kind: str
114 content: str
115 tokens: int | None
116 parent_id: int | None
117 session_id: str | None
118 ts_start: str | None
119 ts_end: str | None
120 created_at: str
121 level: int | None
124@dataclass
125class GrepResult:
126 """A message_event regex match with its covering summary node context (FEAT-1712)."""
128 message_event_id: int
129 session_id: str | None
130 ts: str
131 content: str
132 summary_id: int | None
133 summary_kind: str | None
136@dataclass(frozen=True)
137class SectionProvider:
138 """Config-addressable digest section with query and render logic (ENH-1907)."""
140 name: str
141 query: Callable # (conn, *, cutoff: str, cap: int) -> list
142 default_cap: int
143 render: Callable # (rows: list) -> list[str]
146@dataclass
147class ProjectDigest:
148 """Aggregated project-context snapshot from history.db (ENH-1907)."""
150 sections: list[tuple[str, list[str]]] # [(name, markdown_lines), ...] in config order
151 days: int = 7
153 @property
154 def empty(self) -> bool:
155 return all(not lines for _, lines in self.sections)
158# ---------------------------------------------------------------------------
159# Helpers
160# ---------------------------------------------------------------------------
163def _stale_cutoff(days: int) -> str:
164 """ISO-8601 timestamp *days* ago."""
165 return (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")
168def _connect_readonly(db_path: Path) -> sqlite3.Connection | None:
169 """Open a read-only connection, or return None on failure."""
170 try:
171 ensure_db(db_path)
172 except sqlite3.Error:
173 logger.warning("history_reader: could not ensure schema for %s", db_path, exc_info=True)
174 return None
175 try:
176 conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
177 conn.row_factory = sqlite3.Row
178 conn.execute("PRAGMA query_only = ON")
179 except sqlite3.Error:
180 logger.warning("history_reader: could not open %s read-only", db_path, exc_info=True)
181 return None
182 return conn
185def _row_to_dataclass(row: sqlite3.Row, dc: type[Any]) -> Any:
186 """Map a sqlite3.Row to a dataclass instance, catching extra/unknown keys."""
187 field_names = {f.name for f in dc.__dataclass_fields__.values()}
188 kwargs = {k: row[k] for k in field_names if k in row.keys()}
189 return dc(**kwargs)
192# ---------------------------------------------------------------------------
193# Query API
194# ---------------------------------------------------------------------------
197def find_user_corrections(
198 topic: str,
199 *,
200 limit: int = 10,
201 include_stale: bool = False,
202 db: Path | str = DEFAULT_DB_PATH,
203) -> list[UserCorrection]:
204 """Return user corrections whose content matches *topic* (LIKE search).
206 Stale rows (>30 days by default) are excluded unless *include_stale* is set.
207 """
208 db_path = Path(db)
209 conn = _connect_readonly(db_path)
210 if conn is None:
211 return []
212 try:
213 params: list[Any] = [f"%{topic}%"]
214 where = "WHERE content LIKE ?"
215 if not include_stale:
216 where += " AND ts >= ?"
217 params.append(_stale_cutoff(STALE_DAYS_DEFAULT))
218 rows = conn.execute(
219 f"SELECT ts, session_id, content, source FROM user_corrections {where} "
220 f"ORDER BY ts DESC LIMIT ?",
221 (*params, limit),
222 ).fetchall()
223 except sqlite3.Error:
224 logger.warning("history_reader: find_user_corrections query failed", exc_info=True)
225 return []
226 finally:
227 conn.close()
228 return [_row_to_dataclass(row, UserCorrection) for row in rows]
231def recent_file_events(
232 path: str,
233 *,
234 limit: int = 10,
235 include_stale: bool = False,
236 db: Path | str = DEFAULT_DB_PATH,
237) -> list[FileEvent]:
238 """Return recent file events for *path* (LIKE pattern match).
240 Stale rows (>30 days by default) are excluded unless *include_stale* is set.
241 """
242 db_path = Path(db)
243 conn = _connect_readonly(db_path)
244 if conn is None:
245 return []
246 try:
247 params: list[Any] = [f"%{path}%"]
248 where = "WHERE path LIKE ?"
249 if not include_stale:
250 where += " AND ts >= ?"
251 params.append(_stale_cutoff(STALE_DAYS_DEFAULT))
252 rows = conn.execute(
253 f"SELECT ts, session_id, path, op, issue_id, git_sha FROM file_events {where} "
254 f"ORDER BY ts DESC LIMIT ?",
255 (*params, limit),
256 ).fetchall()
257 except sqlite3.Error:
258 logger.warning("history_reader: recent_file_events query failed", exc_info=True)
259 return []
260 finally:
261 conn.close()
262 return [_row_to_dataclass(row, FileEvent) for row in rows]
265def search(
266 query: str,
267 *,
268 kind: str | None = None,
269 limit: int = 10,
270 db: Path | str = DEFAULT_DB_PATH,
271) -> list[SearchResult]:
272 """FTS5 full-text search with optional *kind* filter (tool, file, issue, loop, correction, message).
274 Returns BM25-ranked results. Gracefully handles invalid FTS5 query syntax.
275 """
276 db_path = Path(db)
277 conn = _connect_readonly(db_path)
278 if conn is None:
279 return []
280 try:
281 if kind:
282 rows = conn.execute(
283 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
284 "FROM search_index WHERE search_index MATCH ? AND kind = ? "
285 "ORDER BY score LIMIT ?",
286 (query, kind, limit),
287 ).fetchall()
288 else:
289 rows = conn.execute(
290 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
291 "FROM search_index WHERE search_index MATCH ? "
292 "ORDER BY score LIMIT ?",
293 (query, limit),
294 ).fetchall()
295 except sqlite3.OperationalError as exc:
296 logger.warning("history_reader: invalid FTS5 query %r: %s", query, exc)
297 return []
298 finally:
299 conn.close()
300 return [_row_to_dataclass(row, SearchResult) for row in rows]
303def related_issue_events(
304 issue_id: str,
305 *,
306 limit: int = 20,
307 db: Path | str = DEFAULT_DB_PATH,
308) -> list[IssueEvent]:
309 """Return issue events for *issue_id*, ordered by most recent first.
311 Searches both the ``issue_events`` table and the FTS5 index for cross-references.
312 """
313 db_path = Path(db)
314 conn = _connect_readonly(db_path)
315 if conn is None:
316 return []
317 try:
318 rows = conn.execute(
319 "SELECT ts, issue_id, transition, discovered_by, issue_type, priority "
320 "FROM issue_events WHERE issue_id = ? "
321 "ORDER BY ts DESC LIMIT ?",
322 (issue_id, limit),
323 ).fetchall()
324 except sqlite3.Error:
325 logger.warning("history_reader: related_issue_events query failed", exc_info=True)
326 return []
327 finally:
328 conn.close()
329 return [_row_to_dataclass(row, IssueEvent) for row in rows]
332def sessions_for_issue(
333 issue_id: str,
334 *,
335 limit: int = 20,
336 db: Path | str = DEFAULT_DB_PATH,
337) -> list[SessionRef]:
338 """Return sessions that co-occurred with *issue_id*'s active period.
340 Queries the ``issue_sessions`` VIEW (v5 migration, ENH-1711), which joins
341 ``issue_events`` to ``message_events`` via overlapping timestamps. Live-emitted
342 rows (from ``issue_lifecycle.py````'s 6 emit sites) now populate ``captured_at``
343 directly; no prior ``backfill`` pass is needed for issues processed after
344 ENH-1839.
346 Returns an empty list when the view is absent (pre-v5 schema), the issue
347 has no recorded sessions, or the database is unavailable.
348 """
349 db_path = Path(db)
350 conn = _connect_readonly(db_path)
351 if conn is None:
352 return []
353 try:
354 rows = conn.execute(
355 "SELECT issue_id, session_id, jsonl_path, first_message_ts, last_message_ts "
356 "FROM issue_sessions WHERE issue_id = ? "
357 "ORDER BY first_message_ts DESC LIMIT ?",
358 (issue_id, limit),
359 ).fetchall()
360 except sqlite3.Error:
361 logger.warning("history_reader: sessions_for_issue query failed", exc_info=True)
362 return []
363 finally:
364 conn.close()
365 return [_row_to_dataclass(row, SessionRef) for row in rows]
368def issue_effort(
369 issue_id: str,
370 *,
371 db: Path | str = DEFAULT_DB_PATH,
372) -> dict | None:
373 """Per-issue effort: session_count and cycle_time_days (first→last session).
375 Returns None when the DB is absent, no sessions exist for the issue, or a
376 query error occurs. Does NOT reuse sessions_for_issue() to avoid the LIMIT=20
377 cap which would produce incorrect cycle_time_days for issues with >20 sessions.
378 """
379 db_path = Path(db)
380 conn = _connect_readonly(db_path)
381 if conn is None:
382 return None
383 try:
384 row = conn.execute(
385 "SELECT COUNT(*) AS session_count, MIN(first_message_ts) AS first_ts, "
386 "MAX(last_message_ts) AS last_ts FROM issue_sessions WHERE issue_id = ?",
387 (issue_id,),
388 ).fetchone()
389 except sqlite3.Error:
390 logger.warning("history_reader: issue_effort query failed", exc_info=True)
391 return None
392 finally:
393 conn.close()
394 if row is None or row["session_count"] == 0:
395 return None
396 cycle: float | None = None
397 if row["first_ts"] and row["last_ts"]:
398 delta = datetime.fromisoformat(row["last_ts"]) - datetime.fromisoformat(row["first_ts"])
399 cycle = delta.total_seconds() / 86400
400 return {"session_count": row["session_count"], "cycle_time_days": cycle}
403def recent_issue_velocity(
404 limit: int = 10,
405 *,
406 db: Path | str = DEFAULT_DB_PATH,
407) -> list[dict]:
408 """Effort data for recently completed issues; empty list when DB has no data.
410 Queries issue_events for recently-completed issues (non-NULL completed_at),
411 then calls issue_effort() for each to produce per-issue effort dicts.
412 """
413 db_path = Path(db)
414 conn = _connect_readonly(db_path)
415 if conn is None:
416 return []
417 try:
418 rows = conn.execute(
419 "SELECT DISTINCT issue_id FROM issue_events "
420 "WHERE completed_at IS NOT NULL "
421 "ORDER BY completed_at DESC LIMIT ?",
422 (limit,),
423 ).fetchall()
424 except sqlite3.Error:
425 logger.warning("history_reader: recent_issue_velocity query failed", exc_info=True)
426 return []
427 finally:
428 conn.close()
429 result = []
430 for row in rows:
431 effort = issue_effort(row["issue_id"], db=db)
432 if effort is not None:
433 result.append({"issue_id": row["issue_id"], **effort})
434 return result
437def lookup_session_metadata(
438 session_id: str,
439 *,
440 db: Path | str = DEFAULT_DB_PATH,
441) -> dict:
442 """Return session-quality metadata dict for a session ID (ENH-1943).
444 Returns:
445 dict with keys: ``has_corrections`` (bool), ``issue_outcome`` (str|None),
446 ``tool_count`` (int), ``files_modified`` (int), ``loop_outcome`` (str|None).
448 ``loop_outcome`` is always ``None`` until ``loop_events`` gains a
449 ``session_id`` column (schema change out of scope).
451 Returns empty dict ``{}`` when DB is missing, empty, or lacks relevant tables.
452 """
453 db_path = Path(db)
454 if not db_path.exists():
455 return {}
456 conn = _connect_readonly(db_path)
457 if conn is None:
458 return {}
459 try:
460 # has_corrections: direct query on user_corrections
461 row = conn.execute(
462 "SELECT COUNT(*) > 0 AS has_corrections FROM user_corrections WHERE session_id = ?",
463 (session_id,),
464 ).fetchone()
465 has_corrections = bool(row["has_corrections"]) if row else False
467 # issue_outcome: JOIN through issue_sessions VIEW (issue_events has no
468 # session_id column; migration v5 bridges via the VIEW)
469 row = conn.execute(
470 "SELECT ie.transition "
471 "FROM issue_sessions is2 "
472 "JOIN issue_events ie ON is2.issue_id = ie.issue_id "
473 "WHERE is2.session_id = ? AND ie.transition = 'done' "
474 "ORDER BY ie.ts DESC LIMIT 1",
475 (session_id,),
476 ).fetchone()
477 issue_outcome: str | None = row["transition"] if row else None
479 # tool_count: direct query on tool_events
480 row = conn.execute(
481 "SELECT COUNT(*) AS tool_count FROM tool_events WHERE session_id = ?",
482 (session_id,),
483 ).fetchone()
484 tool_count: int = row["tool_count"] if row else 0
486 # files_modified: direct query on file_events; op values include both
487 # hook-written tool names ('Write') and lowercase variants ('write', 'create')
488 row = conn.execute(
489 "SELECT COUNT(*) AS files_modified FROM file_events "
490 "WHERE session_id = ? AND op IN ('write', 'create', 'Write')",
491 (session_id,),
492 ).fetchone()
493 files_modified: int = row["files_modified"] if row else 0
495 # loop_outcome: loop_events has no session_id column; always None
496 loop_outcome: None = None
498 except sqlite3.Error:
499 logger.warning("history_reader: lookup_session_metadata query failed", exc_info=True)
500 return {}
501 finally:
502 conn.close()
504 return {
505 "has_corrections": has_corrections,
506 "issue_outcome": issue_outcome,
507 "tool_count": tool_count,
508 "files_modified": files_modified,
509 "loop_outcome": loop_outcome,
510 }
513def conversation_turns(
514 db_path: Path | str,
515 since: datetime | None = None,
516 context_window: int = 3,
517) -> list[list[tuple[str, str]]]:
518 """Return conversation turn-pair windows from ``history.db`` (ENH-1942).
520 Queries ``message_events`` and ``assistant_messages``, pairs user messages
521 with their assistant responses via temporal adjacency (same algorithm as
522 ``_extract_turn_pairs()`` in ``user_messages.py``), and groups them into
523 sliding windows of *context_window* turn-pairs each.
525 Returns ``[]`` when the database is missing, empty, predates schema v11
526 (no ``assistant_messages`` table), or when no turn-pairs match the *since*
527 filter. Callers should fall back to JSONL parsing in that case.
529 Args:
530 db_path: Path to ``history.db``.
531 since: Only include turns where the user message timestamp is >= this value.
532 context_window: Number of (user, assistant) turn-pairs per output window.
534 Returns:
535 List of conversation windows; each window is a ``list[tuple[str, str]]``
536 alternating between ``("user", text)`` and ``("assistant", text)``.
537 """
538 db_path = Path(db_path)
539 if not db_path.exists():
540 return []
541 conn = _connect_readonly(db_path)
542 if conn is None:
543 return []
544 try:
545 # Check that assistant_messages table exists (schema >= v11)
546 row = conn.execute(
547 "SELECT COUNT(*) AS n FROM sqlite_master "
548 "WHERE type = 'table' AND name = 'assistant_messages'"
549 ).fetchone()
550 if not row or row["n"] == 0:
551 return []
553 # Pair each user message with assistant messages between it and the
554 # NEXT user message (temporal-adjacency, matching _extract_turn_pairs).
555 # The subquery finds the next user message timestamp per session;
556 # COALESCE defaults to a far-future sentinel for the last message.
557 base_sql = (
558 "SELECT u.session_id, u.ts AS user_ts, u.content AS user_text, "
559 "a.content AS assistant_text "
560 "FROM message_events u "
561 "JOIN assistant_messages a ON a.session_id = u.session_id "
562 "AND a.ts > u.ts "
563 "AND a.ts < COALESCE("
564 " (SELECT MIN(u2.ts) FROM message_events u2 "
565 " WHERE u2.session_id = u.session_id AND u2.ts > u.ts), "
566 " '9999-12-31'"
567 ") "
568 )
569 params: list[Any] = []
570 if since is not None:
571 base_sql += "WHERE u.ts >= ? "
572 params.append(since.strftime("%Y-%m-%dT%H:%M:%SZ"))
573 base_sql += "ORDER BY u.session_id, u.ts, a.ts"
575 rows = conn.execute(base_sql, params or ()).fetchall()
577 if not rows:
578 return []
580 # Group assistant texts by user message.
581 # The SQL already only includes assistant messages between consecutive
582 # user messages, so simple grouping by (session_id, user_ts) suffices.
583 turn_pairs: list[tuple[str, str]] = []
584 current_key: tuple[str, str] | None = None
585 current_user: str = ""
586 assistant_texts: list[str] = []
588 for row_ in rows:
589 key = (row_["session_id"], row_["user_ts"])
590 if key != current_key:
591 if current_key is not None and assistant_texts:
592 turn_pairs.append((current_user, "\n\n".join(assistant_texts)))
593 current_key = key
594 current_user = row_["user_text"]
595 assistant_texts = []
596 assistant_texts.append(row_["assistant_text"])
598 # Flush the final turn
599 if current_key is not None and assistant_texts:
600 turn_pairs.append((current_user, "\n\n".join(assistant_texts)))
602 # Emit sliding windows of context_window turn-pairs
603 windows: list[list[tuple[str, str]]] = []
604 n = len(turn_pairs)
605 if n == 0:
606 return []
607 for i in range(max(1, n - context_window + 1)):
608 window_pairs = turn_pairs[i : i + context_window]
609 window: list[tuple[str, str]] = []
610 for user_text, assistant_text in window_pairs:
611 window.append(("user", user_text))
612 window.append(("assistant", assistant_text))
613 windows.append(window)
615 return windows
617 except sqlite3.Error:
618 logger.warning("history_reader: conversation_turns query failed", exc_info=True)
619 return []
620 finally:
621 conn.close()
624# ---------------------------------------------------------------------------
625# Summary DAG retrieval (FEAT-1712)
626# ---------------------------------------------------------------------------
629def ll_grep(
630 pattern: str,
631 *,
632 summary_id: int | None = None,
633 limit: int = 50,
634 db: Path | str = DEFAULT_DB_PATH,
635) -> list[GrepResult]:
636 """Regex search over message_events, with covering summary node context.
638 Each result includes the summary_id and summary_kind of the leaf node covering the
639 matched message (or None/None for messages not yet compacted). If *summary_id* is
640 provided, restrict the search to messages covered by that specific node.
642 When *summary_id* is a condensed node (kind='condensed'), uses a recursive CTE to
643 walk the N-level DAG (condensed → … → leaves via parent_id → message_events via
644 summary_spans) so that messages under all descendant leaves are searched regardless
645 of condensation depth.
646 """
647 db_path = Path(db)
648 conn = _connect_readonly(db_path)
649 if conn is None:
650 return []
652 def _regexp(pat: str, val: str | None) -> bool:
653 try:
654 return bool(re.search(pat, val or "", re.IGNORECASE))
655 except re.error:
656 return False
658 try:
659 conn.create_function("regexp_match", 2, _regexp)
660 if summary_id is not None:
661 # Recursive CTE walks the full N-level DAG from the starting node
662 # through all descendants, terminating at leaf nodes that have
663 # summary_spans entries. Works uniformly for both kind='leaf'
664 # (CTE = 1 row) and kind='condensed' at any depth.
665 rows = conn.execute(
666 "WITH RECURSIVE descendants AS ("
667 " SELECT id, kind FROM summary_nodes WHERE id = ?1"
668 " UNION ALL"
669 " SELECT sn.id, sn.kind"
670 " FROM summary_nodes sn"
671 " JOIN descendants d ON sn.parent_id = d.id"
672 ")"
673 "SELECT me.id, me.session_id, me.ts, me.content,"
674 " sn.id AS summary_id, sn.kind AS summary_kind"
675 " FROM message_events me"
676 " JOIN summary_spans ss ON ss.message_event_id = me.id"
677 " JOIN descendants leaf ON leaf.id = ss.summary_id"
678 " JOIN summary_nodes sn ON sn.id = leaf.id"
679 " WHERE regexp_match(?2, me.content)"
680 " ORDER BY me.ts, me.id LIMIT ?3",
681 (summary_id, pattern, limit),
682 ).fetchall()
683 else:
684 rows = conn.execute(
685 "SELECT me.id, me.session_id, me.ts, me.content,"
686 " sn.id AS summary_id, sn.kind AS summary_kind"
687 " FROM message_events me"
688 " LEFT JOIN summary_spans ss ON ss.message_event_id = me.id"
689 " LEFT JOIN summary_nodes sn ON sn.id = ss.summary_id"
690 " WHERE regexp_match(?, me.content)"
691 " ORDER BY me.ts, me.id LIMIT ?",
692 (pattern, limit),
693 ).fetchall()
694 except sqlite3.Error:
695 logger.warning("history_reader: ll_grep query failed", exc_info=True)
696 return []
697 finally:
698 conn.close()
700 return [
701 GrepResult(
702 message_event_id=row["id"],
703 session_id=row["session_id"],
704 ts=row["ts"],
705 content=row["content"] or "",
706 summary_id=row["summary_id"],
707 summary_kind=row["summary_kind"],
708 )
709 for row in rows
710 ]
713def ll_expand(
714 summary_id: int,
715 *,
716 db: Path | str = DEFAULT_DB_PATH,
717) -> list[dict]:
718 """Return the message_events covered by *summary_id*.
720 Uses a recursive CTE to walk the N-level summary DAG from the starting
721 node through all descendants, terminating at leaf nodes that have
722 ``summary_spans`` entries. Works uniformly for both ``kind='leaf'``
723 (CTE = 1 row, direct span join) and ``kind='condensed'`` at any
724 condensation depth (CTE descends through intermediate condensed nodes
725 to reach the leaves).
727 Returns dicts with keys ``id``, ``session_id``, ``ts``, ``content``.
728 Empty list when the summary node does not exist or has no spans.
729 """
730 db_path = Path(db)
731 conn = _connect_readonly(db_path)
732 if conn is None:
733 return []
734 try:
735 # Recursive CTE handles leaf and condensed nodes uniformly at any depth.
736 # For a leaf node, descendants = {itself} and the summary_spans join is direct.
737 # For a condensed node, descendants = {condensed, children, grandchildren, …}
738 # and the summary_spans join only matches the leaf-descendant rows.
739 rows = conn.execute(
740 "WITH RECURSIVE descendants AS ("
741 " SELECT id, kind FROM summary_nodes WHERE id = ?1"
742 " UNION ALL"
743 " SELECT sn.id, sn.kind"
744 " FROM summary_nodes sn"
745 " JOIN descendants d ON sn.parent_id = d.id"
746 ")"
747 "SELECT me.id, me.session_id, me.ts, me.content"
748 " FROM message_events me"
749 " JOIN summary_spans ss ON ss.message_event_id = me.id"
750 " JOIN descendants leaf ON leaf.id = ss.summary_id"
751 " ORDER BY me.ts, me.id",
752 (summary_id,),
753 ).fetchall()
754 except sqlite3.Error:
755 logger.warning("history_reader: ll_expand query failed", exc_info=True)
756 return []
757 finally:
758 conn.close()
759 return [dict(row) for row in rows]
762def ll_describe(
763 node_id: int,
764 *,
765 db: Path | str = DEFAULT_DB_PATH,
766) -> SummaryNode | None:
767 """Return metadata for a summary_nodes row.
769 Returns ``None`` when the node does not exist or the database is unavailable.
770 """
771 db_path = Path(db)
772 conn = _connect_readonly(db_path)
773 if conn is None:
774 return None
775 try:
776 row = conn.execute(
777 "SELECT id, kind, content, tokens, parent_id, session_id,"
778 " ts_start, ts_end, created_at, level"
779 " FROM summary_nodes WHERE id = ?",
780 (node_id,),
781 ).fetchone()
782 except sqlite3.Error:
783 logger.warning("history_reader: ll_describe query failed", exc_info=True)
784 return None
785 finally:
786 conn.close()
787 if row is None:
788 return None
789 return SummaryNode(
790 id=row["id"],
791 kind=row["kind"],
792 content=row["content"],
793 tokens=row["tokens"],
794 parent_id=row["parent_id"],
795 session_id=row["session_id"],
796 ts_start=row["ts_start"],
797 ts_end=row["ts_end"],
798 created_at=row["created_at"],
799 level=row["level"],
800 )
803def condensed_nodes_for_issue(
804 issue_id: str,
805 *,
806 limit: int = 3,
807 node_char_cap: int = 500,
808 db: Path | str = DEFAULT_DB_PATH,
809) -> list[SummaryNode]:
810 """Return level-0 condensed summary_nodes for an issue's sessions (ENH-2231).
812 Joins the ``issue_sessions`` VIEW to ``summary_nodes`` filtering for
813 ``kind='condensed'`` and ``level=0`` (per-session condensed nodes, one per
814 session). Returns nodes newest-first, limited to *limit*. Each node's
815 ``content`` is truncated to *node_char_cap* characters.
817 Returns an empty list when the DB is absent, the issue has no recorded
818 sessions, no condensed nodes have been generated, or any query error occurs.
819 """
820 db_path = Path(db)
821 conn = _connect_readonly(db_path)
822 if conn is None:
823 return []
824 try:
825 rows = conn.execute(
826 "SELECT sn.id, sn.kind, sn.content, sn.tokens, sn.parent_id, sn.session_id,"
827 " sn.ts_start, sn.ts_end, sn.created_at, sn.level"
828 " FROM summary_nodes sn"
829 " JOIN issue_sessions isl ON isl.session_id = sn.session_id"
830 " WHERE isl.issue_id = ?"
831 " AND sn.kind = 'condensed'"
832 " AND sn.level = 0"
833 " ORDER BY sn.ts_end DESC, sn.id DESC"
834 " LIMIT ?",
835 (issue_id, limit),
836 ).fetchall()
837 except sqlite3.Error:
838 logger.warning("history_reader: condensed_nodes_for_issue query failed", exc_info=True)
839 return []
840 finally:
841 conn.close()
842 return [
843 SummaryNode(
844 id=row["id"],
845 kind=row["kind"],
846 content=(row["content"] or "")[:node_char_cap],
847 tokens=row["tokens"],
848 parent_id=row["parent_id"],
849 session_id=row["session_id"],
850 ts_start=row["ts_start"],
851 ts_end=row["ts_end"],
852 created_at=row["created_at"],
853 level=row["level"],
854 )
855 for row in rows
856 ]
859# ---------------------------------------------------------------------------
860# Project digest — section providers (ENH-1907)
861# ---------------------------------------------------------------------------
864def _query_touched_files(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
865 try:
866 return conn.execute(
867 "SELECT path, COUNT(*) AS edit_count "
868 "FROM file_events "
869 "WHERE ts >= ? AND path IS NOT NULL "
870 "GROUP BY path "
871 "ORDER BY edit_count DESC, MAX(ts) DESC "
872 "LIMIT ?",
873 (cutoff, cap),
874 ).fetchall()
875 except sqlite3.Error:
876 return []
879def _render_touched_files(rows: list) -> list[str]:
880 lines = []
881 for row in rows:
882 path = row["path"]
883 count = row["edit_count"]
884 noun = "edit" if count == 1 else "edits"
885 lines.append(f"- {path} ({count} {noun})")
886 return lines
889def _query_completed_issues(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
890 try:
891 return conn.execute(
892 "SELECT ts, issue_id, transition, issue_type, priority "
893 "FROM issue_events "
894 "WHERE transition IN ('done', 'cancelled') AND ts >= ? "
895 "ORDER BY ts DESC "
896 "LIMIT ?",
897 (cutoff, cap),
898 ).fetchall()
899 except sqlite3.Error:
900 return []
903def _render_completed_issues(rows: list) -> list[str]:
904 lines = []
905 now = datetime.now(UTC)
906 for row in rows:
907 issue_id = row["issue_id"] or "unknown"
908 ts_str = row["ts"] or ""
909 time_ago = ""
910 if ts_str:
911 try:
912 ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
913 delta_days = (now - ts).days
914 if delta_days == 0:
915 time_ago = " (today)"
916 elif delta_days == 1:
917 time_ago = " (1d ago)"
918 else:
919 time_ago = f" ({delta_days}d ago)"
920 except (ValueError, TypeError):
921 pass
922 lines.append(f"- {issue_id}{time_ago}")
923 return lines
926def _query_recurring_corrections(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
927 try:
928 return conn.execute(
929 "SELECT content, COUNT(*) AS seen_count "
930 "FROM user_corrections "
931 "WHERE ts >= ? "
932 "GROUP BY content "
933 "ORDER BY seen_count DESC, MAX(ts) DESC "
934 "LIMIT ?",
935 (cutoff, cap),
936 ).fetchall()
937 except sqlite3.Error:
938 return []
941def _render_recurring_corrections(rows: list) -> list[str]:
942 lines = []
943 for row in rows:
944 content = row["content"] or ""
945 count = row["seen_count"]
946 if len(content) > 80:
947 content = content[:77] + "..."
948 count_str = f" (seen {count}x)" if count > 1 else ""
949 lines.append(f'- "{content}"{count_str}')
950 return lines
953SECTION_PROVIDERS: dict[str, SectionProvider] = {
954 "touched_files": SectionProvider(
955 name="touched_files",
956 query=_query_touched_files,
957 default_cap=10,
958 render=_render_touched_files,
959 ),
960 "completed_issues": SectionProvider(
961 name="completed_issues",
962 query=_query_completed_issues,
963 default_cap=5,
964 render=_render_completed_issues,
965 ),
966 "recurring_corrections": SectionProvider(
967 name="recurring_corrections",
968 query=_query_recurring_corrections,
969 default_cap=5,
970 render=_render_recurring_corrections,
971 ),
972}
974_SECTION_HEADERS: dict[str, str] = {
975 "touched_files": "Recently touched (last {days} days)",
976 "completed_issues": "Recently completed issues",
977 "recurring_corrections": "Recurring corrections",
978}
981def project_digest(
982 db_path: Path,
983 *,
984 days: int = 7,
985 sections: list[str] | None = None,
986) -> ProjectDigest:
987 """Aggregate a project-wide context snapshot from history.db.
989 Returns a :class:`ProjectDigest` with ``.empty == True`` on missing /
990 empty / stale DB. ``sections=None`` or ``sections=[]`` renders all
991 registered providers in registry order; a non-empty list restricts and
992 orders the output.
993 """
994 conn = _connect_readonly(db_path)
995 if conn is None:
996 return ProjectDigest(sections=[], days=days)
998 cutoff = _stale_cutoff(days)
999 provider_keys: list[str] = list(SECTION_PROVIDERS.keys()) if sections is None else sections
1001 result: list[tuple[str, list[str]]] = []
1002 try:
1003 for key in provider_keys:
1004 provider = SECTION_PROVIDERS.get(key)
1005 if provider is None:
1006 logger.warning("project_digest: unknown section %r — skipping", key)
1007 continue
1008 rows = provider.query(conn, cutoff=cutoff, cap=provider.default_cap)
1009 lines = provider.render(rows) if rows else []
1010 if lines:
1011 result.append((key, lines))
1012 finally:
1013 conn.close()
1015 return ProjectDigest(sections=result, days=days)
1018def render_project_context(
1019 digest: ProjectDigest,
1020 *,
1021 char_cap: int = 1200,
1022 days: int | None = None,
1023) -> str:
1024 """Render a ``<project_context>`` block from *digest*, capped at *char_cap* chars.
1026 Returns ``""`` when the digest is empty (no block injected). Truncates
1027 with a ``+N more`` tail when content would exceed *char_cap*.
1028 """
1029 if digest.empty:
1030 return ""
1032 effective_days = days if days is not None else digest.days
1033 content_lines: list[str] = []
1034 for name, section_lines in digest.sections:
1035 raw_header = _SECTION_HEADERS.get(name, name.replace("_", " ").title())
1036 header = raw_header.format(days=effective_days)
1037 content_lines.append(f"## {header}")
1038 content_lines.extend(section_lines)
1040 open_tag = "<project_context>"
1041 close_tag = "</project_context>"
1043 full_block = "\n".join([open_tag] + content_lines + [close_tag])
1044 if len(full_block) <= char_cap:
1045 return full_block
1047 # Truncate: accumulate content lines under budget, append "+N more" tail.
1048 close_cost = len("\n" + close_tag)
1049 budget = char_cap - len(open_tag) - close_cost - 1 # -1 for opening newline
1050 accepted: list[str] = []
1051 dropped = 0
1052 for line in content_lines:
1053 line_cost = len(line) + 1 # +1 for the newline separator
1054 if budget >= line_cost:
1055 accepted.append(line)
1056 budget -= line_cost
1057 else:
1058 dropped += 1
1060 if dropped:
1061 tail = f"... +{dropped} more"
1062 if budget >= len(tail) + 1:
1063 accepted.append(tail)
1065 return "\n".join([open_tag] + accepted + [close_tag])