Coverage for little_loops / history_reader.py: 28%
366 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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 ll_grep(pattern, ...) -> list[GrepResult]
27 ll_expand(summary_id, ...) -> list[dict]
28 ll_describe(node_id, ...) -> SummaryNode | None
29 project_digest(db_path, ...) -> ProjectDigest
30 render_project_context(digest, ...) -> str
31"""
33from __future__ import annotations
35import logging
36import re
37import sqlite3
38from collections.abc import Callable
39from dataclasses import dataclass
40from datetime import UTC, datetime, timedelta
41from pathlib import Path
42from typing import Any
44from little_loops.session_store import DEFAULT_DB_PATH, ensure_db
46logger = logging.getLogger(__name__)
48STALE_DAYS_DEFAULT = 30
51# ---------------------------------------------------------------------------
52# Dataclasses
53# ---------------------------------------------------------------------------
56@dataclass
57class UserCorrection:
58 ts: str
59 session_id: str | None
60 content: str
61 source: str | None
64@dataclass
65class FileEvent:
66 ts: str
67 session_id: str | None
68 path: str | None
69 op: str | None
70 issue_id: str | None
71 git_sha: str | None
74@dataclass
75class SearchResult:
76 content: str
77 kind: str
78 ref: str
79 anchor: str
80 ts: str
81 score: float
84@dataclass
85class IssueEvent:
86 ts: str
87 issue_id: str | None
88 transition: str | None
89 discovered_by: str | None
90 issue_type: str | None
91 priority: str | None
94@dataclass
95class SessionRef:
96 """A session that co-occurred with an issue's active period (ENH-1711)."""
98 issue_id: str | None
99 session_id: str | None
100 jsonl_path: str | None
101 first_message_ts: str | None
102 last_message_ts: str | None
105@dataclass
106class SummaryNode:
107 """A summary_nodes row from the LCM-style compaction DAG (FEAT-1712)."""
109 id: int
110 kind: str
111 content: str
112 tokens: int | None
113 parent_id: int | None
114 session_id: str | None
115 ts_start: str | None
116 ts_end: str | None
117 created_at: str
120@dataclass
121class GrepResult:
122 """A message_event regex match with its covering summary node context (FEAT-1712)."""
124 message_event_id: int
125 session_id: str | None
126 ts: str
127 content: str
128 summary_id: int | None
129 summary_kind: str | None
132@dataclass(frozen=True)
133class SectionProvider:
134 """Config-addressable digest section with query and render logic (ENH-1907)."""
136 name: str
137 query: Callable # (conn, *, cutoff: str, cap: int) -> list
138 default_cap: int
139 render: Callable # (rows: list) -> list[str]
142@dataclass
143class ProjectDigest:
144 """Aggregated project-context snapshot from history.db (ENH-1907)."""
146 sections: list[tuple[str, list[str]]] # [(name, markdown_lines), ...] in config order
147 days: int = 7
149 @property
150 def empty(self) -> bool:
151 return all(not lines for _, lines in self.sections)
154# ---------------------------------------------------------------------------
155# Helpers
156# ---------------------------------------------------------------------------
159def _stale_cutoff(days: int) -> str:
160 """ISO-8601 timestamp *days* ago."""
161 return (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")
164def _connect_readonly(db_path: Path) -> sqlite3.Connection | None:
165 """Open a read-only connection, or return None on failure."""
166 try:
167 ensure_db(db_path)
168 except sqlite3.Error:
169 logger.warning("history_reader: could not ensure schema for %s", db_path, exc_info=True)
170 return None
171 try:
172 conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
173 conn.row_factory = sqlite3.Row
174 conn.execute("PRAGMA query_only = ON")
175 except sqlite3.Error:
176 logger.warning("history_reader: could not open %s read-only", db_path, exc_info=True)
177 return None
178 return conn
181def _row_to_dataclass(row: sqlite3.Row, dc: type[Any]) -> Any:
182 """Map a sqlite3.Row to a dataclass instance, catching extra/unknown keys."""
183 field_names = {f.name for f in dc.__dataclass_fields__.values()}
184 kwargs = {k: row[k] for k in field_names if k in row.keys()}
185 return dc(**kwargs)
188# ---------------------------------------------------------------------------
189# Query API
190# ---------------------------------------------------------------------------
193def find_user_corrections(
194 topic: str,
195 *,
196 limit: int = 10,
197 include_stale: bool = False,
198 db: Path | str = DEFAULT_DB_PATH,
199) -> list[UserCorrection]:
200 """Return user corrections whose content matches *topic* (LIKE search).
202 Stale rows (>30 days by default) are excluded unless *include_stale* is set.
203 """
204 db_path = Path(db)
205 conn = _connect_readonly(db_path)
206 if conn is None:
207 return []
208 try:
209 params: list[Any] = [f"%{topic}%"]
210 where = "WHERE content LIKE ?"
211 if not include_stale:
212 where += " AND ts >= ?"
213 params.append(_stale_cutoff(STALE_DAYS_DEFAULT))
214 rows = conn.execute(
215 f"SELECT ts, session_id, content, source FROM user_corrections {where} "
216 f"ORDER BY ts DESC LIMIT ?",
217 (*params, limit),
218 ).fetchall()
219 except sqlite3.Error:
220 logger.warning("history_reader: find_user_corrections query failed", exc_info=True)
221 return []
222 finally:
223 conn.close()
224 return [_row_to_dataclass(row, UserCorrection) for row in rows]
227def recent_file_events(
228 path: str,
229 *,
230 limit: int = 10,
231 include_stale: bool = False,
232 db: Path | str = DEFAULT_DB_PATH,
233) -> list[FileEvent]:
234 """Return recent file events for *path* (LIKE pattern match).
236 Stale rows (>30 days by default) are excluded unless *include_stale* is set.
237 """
238 db_path = Path(db)
239 conn = _connect_readonly(db_path)
240 if conn is None:
241 return []
242 try:
243 params: list[Any] = [f"%{path}%"]
244 where = "WHERE path LIKE ?"
245 if not include_stale:
246 where += " AND ts >= ?"
247 params.append(_stale_cutoff(STALE_DAYS_DEFAULT))
248 rows = conn.execute(
249 f"SELECT ts, session_id, path, op, issue_id, git_sha FROM file_events {where} "
250 f"ORDER BY ts DESC LIMIT ?",
251 (*params, limit),
252 ).fetchall()
253 except sqlite3.Error:
254 logger.warning("history_reader: recent_file_events query failed", exc_info=True)
255 return []
256 finally:
257 conn.close()
258 return [_row_to_dataclass(row, FileEvent) for row in rows]
261def search(
262 query: str,
263 *,
264 kind: str | None = None,
265 limit: int = 10,
266 db: Path | str = DEFAULT_DB_PATH,
267) -> list[SearchResult]:
268 """FTS5 full-text search with optional *kind* filter (tool, file, issue, loop, correction, message).
270 Returns BM25-ranked results. Gracefully handles invalid FTS5 query syntax.
271 """
272 db_path = Path(db)
273 conn = _connect_readonly(db_path)
274 if conn is None:
275 return []
276 try:
277 if kind:
278 rows = conn.execute(
279 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
280 "FROM search_index WHERE search_index MATCH ? AND kind = ? "
281 "ORDER BY score LIMIT ?",
282 (query, kind, limit),
283 ).fetchall()
284 else:
285 rows = conn.execute(
286 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
287 "FROM search_index WHERE search_index MATCH ? "
288 "ORDER BY score LIMIT ?",
289 (query, limit),
290 ).fetchall()
291 except sqlite3.OperationalError as exc:
292 logger.warning("history_reader: invalid FTS5 query %r: %s", query, exc)
293 return []
294 finally:
295 conn.close()
296 return [_row_to_dataclass(row, SearchResult) for row in rows]
299def related_issue_events(
300 issue_id: str,
301 *,
302 limit: int = 20,
303 db: Path | str = DEFAULT_DB_PATH,
304) -> list[IssueEvent]:
305 """Return issue events for *issue_id*, ordered by most recent first.
307 Searches both the ``issue_events`` table and the FTS5 index for cross-references.
308 """
309 db_path = Path(db)
310 conn = _connect_readonly(db_path)
311 if conn is None:
312 return []
313 try:
314 rows = conn.execute(
315 "SELECT ts, issue_id, transition, discovered_by, issue_type, priority "
316 "FROM issue_events WHERE issue_id = ? "
317 "ORDER BY ts DESC LIMIT ?",
318 (issue_id, limit),
319 ).fetchall()
320 except sqlite3.Error:
321 logger.warning("history_reader: related_issue_events query failed", exc_info=True)
322 return []
323 finally:
324 conn.close()
325 return [_row_to_dataclass(row, IssueEvent) for row in rows]
328def sessions_for_issue(
329 issue_id: str,
330 *,
331 limit: int = 20,
332 db: Path | str = DEFAULT_DB_PATH,
333) -> list[SessionRef]:
334 """Return sessions that co-occurred with *issue_id*'s active period.
336 Queries the ``issue_sessions`` VIEW (v5 migration, ENH-1711), which joins
337 ``issue_events`` to ``message_events`` via overlapping timestamps. Live-emitted
338 rows (from ``issue_lifecycle.py````'s 6 emit sites) now populate ``captured_at``
339 directly; no prior ``backfill`` pass is needed for issues processed after
340 ENH-1839.
342 Returns an empty list when the view is absent (pre-v5 schema), the issue
343 has no recorded sessions, or the database is unavailable.
344 """
345 db_path = Path(db)
346 conn = _connect_readonly(db_path)
347 if conn is None:
348 return []
349 try:
350 rows = conn.execute(
351 "SELECT issue_id, session_id, jsonl_path, first_message_ts, last_message_ts "
352 "FROM issue_sessions WHERE issue_id = ? "
353 "ORDER BY first_message_ts DESC LIMIT ?",
354 (issue_id, limit),
355 ).fetchall()
356 except sqlite3.Error:
357 logger.warning("history_reader: sessions_for_issue query failed", exc_info=True)
358 return []
359 finally:
360 conn.close()
361 return [_row_to_dataclass(row, SessionRef) for row in rows]
364def issue_effort(
365 issue_id: str,
366 *,
367 db: Path | str = DEFAULT_DB_PATH,
368) -> dict | None:
369 """Per-issue effort: session_count and cycle_time_days (first→last session).
371 Returns None when the DB is absent, no sessions exist for the issue, or a
372 query error occurs. Does NOT reuse sessions_for_issue() to avoid the LIMIT=20
373 cap which would produce incorrect cycle_time_days for issues with >20 sessions.
374 """
375 db_path = Path(db)
376 conn = _connect_readonly(db_path)
377 if conn is None:
378 return None
379 try:
380 row = conn.execute(
381 "SELECT COUNT(*) AS session_count, MIN(first_message_ts) AS first_ts, "
382 "MAX(last_message_ts) AS last_ts FROM issue_sessions WHERE issue_id = ?",
383 (issue_id,),
384 ).fetchone()
385 except sqlite3.Error:
386 logger.warning("history_reader: issue_effort query failed", exc_info=True)
387 return None
388 finally:
389 conn.close()
390 if row is None or row["session_count"] == 0:
391 return None
392 cycle: float | None = None
393 if row["first_ts"] and row["last_ts"]:
394 delta = datetime.fromisoformat(row["last_ts"]) - datetime.fromisoformat(row["first_ts"])
395 cycle = delta.total_seconds() / 86400
396 return {"session_count": row["session_count"], "cycle_time_days": cycle}
399def recent_issue_velocity(
400 limit: int = 10,
401 *,
402 db: Path | str = DEFAULT_DB_PATH,
403) -> list[dict]:
404 """Effort data for recently completed issues; empty list when DB has no data.
406 Queries issue_events for recently-completed issues (non-NULL completed_at),
407 then calls issue_effort() for each to produce per-issue effort dicts.
408 """
409 db_path = Path(db)
410 conn = _connect_readonly(db_path)
411 if conn is None:
412 return []
413 try:
414 rows = conn.execute(
415 "SELECT DISTINCT issue_id FROM issue_events "
416 "WHERE completed_at IS NOT NULL "
417 "ORDER BY completed_at DESC LIMIT ?",
418 (limit,),
419 ).fetchall()
420 except sqlite3.Error:
421 logger.warning("history_reader: recent_issue_velocity query failed", exc_info=True)
422 return []
423 finally:
424 conn.close()
425 result = []
426 for row in rows:
427 effort = issue_effort(row["issue_id"], db=db)
428 if effort is not None:
429 result.append({"issue_id": row["issue_id"], **effort})
430 return result
433# ---------------------------------------------------------------------------
434# Summary DAG retrieval (FEAT-1712)
435# ---------------------------------------------------------------------------
438def ll_grep(
439 pattern: str,
440 *,
441 summary_id: int | None = None,
442 limit: int = 50,
443 db: Path | str = DEFAULT_DB_PATH,
444) -> list[GrepResult]:
445 """Regex search over message_events, with covering summary node context.
447 Each result includes the summary_id and summary_kind of the leaf node covering the
448 matched message (or None/None for messages not yet compacted). If *summary_id* is
449 provided, restrict the search to messages covered by that specific node.
451 When *summary_id* is a condensed node (kind='condensed'), uses a two-hop traversal
452 (condensed → leaves via parent_id → message_events via summary_spans) so that
453 messages under all constituent leaves are searched.
454 """
455 db_path = Path(db)
456 conn = _connect_readonly(db_path)
457 if conn is None:
458 return []
460 def _regexp(pat: str, val: str | None) -> bool:
461 try:
462 return bool(re.search(pat, val or "", re.IGNORECASE))
463 except re.error:
464 return False
466 try:
467 conn.create_function("regexp_match", 2, _regexp)
468 if summary_id is not None:
469 # Check if this is a condensed node for two-hop traversal
470 kind_row = conn.execute(
471 "SELECT kind FROM summary_nodes WHERE id = ?", (summary_id,)
472 ).fetchone()
473 if kind_row and kind_row["kind"] == "condensed":
474 rows = conn.execute(
475 "SELECT me.id, me.session_id, me.ts, me.content,"
476 " sn.id AS summary_id, sn.kind AS summary_kind"
477 " FROM message_events me"
478 " JOIN summary_spans ss ON ss.message_event_id = me.id"
479 " JOIN summary_nodes leaf ON leaf.id = ss.summary_id"
480 " JOIN summary_nodes sn ON sn.id = leaf.id"
481 " WHERE leaf.parent_id = ? AND regexp_match(?, me.content)"
482 " ORDER BY me.ts, me.id LIMIT ?",
483 (summary_id, pattern, limit),
484 ).fetchall()
485 else:
486 rows = conn.execute(
487 "SELECT me.id, me.session_id, me.ts, me.content,"
488 " sn.id AS summary_id, sn.kind AS summary_kind"
489 " FROM message_events me"
490 " JOIN summary_spans ss ON ss.message_event_id = me.id"
491 " JOIN summary_nodes sn ON sn.id = ss.summary_id"
492 " WHERE ss.summary_id = ? AND regexp_match(?, me.content)"
493 " ORDER BY me.ts, me.id LIMIT ?",
494 (summary_id, pattern, limit),
495 ).fetchall()
496 else:
497 rows = conn.execute(
498 "SELECT me.id, me.session_id, me.ts, me.content,"
499 " sn.id AS summary_id, sn.kind AS summary_kind"
500 " FROM message_events me"
501 " LEFT JOIN summary_spans ss ON ss.message_event_id = me.id"
502 " LEFT JOIN summary_nodes sn ON sn.id = ss.summary_id"
503 " WHERE regexp_match(?, me.content)"
504 " ORDER BY me.ts, me.id LIMIT ?",
505 (pattern, limit),
506 ).fetchall()
507 except sqlite3.Error:
508 logger.warning("history_reader: ll_grep query failed", exc_info=True)
509 return []
510 finally:
511 conn.close()
513 return [
514 GrepResult(
515 message_event_id=row["id"],
516 session_id=row["session_id"],
517 ts=row["ts"],
518 content=row["content"] or "",
519 summary_id=row["summary_id"],
520 summary_kind=row["summary_kind"],
521 )
522 for row in rows
523 ]
526def ll_expand(
527 summary_id: int,
528 *,
529 db: Path | str = DEFAULT_DB_PATH,
530) -> list[dict]:
531 """Return the message_events covered by *summary_id*.
533 For ``kind='leaf'`` nodes, traverses ``summary_spans`` directly to
534 message_events. For ``kind='condensed'`` nodes, traverses a two-hop
535 path: ``condensed → leaves (via parent_id) → message_events (via
536 summary_spans)``.
538 Returns dicts with keys ``id``, ``session_id``, ``ts``, ``content``.
539 Empty list when the summary node does not exist or has no spans.
540 """
541 db_path = Path(db)
542 conn = _connect_readonly(db_path)
543 if conn is None:
544 return []
545 try:
546 # Check node kind to choose the correct traversal
547 kind_row = conn.execute(
548 "SELECT kind FROM summary_nodes WHERE id = ?", (summary_id,)
549 ).fetchone()
550 if kind_row is None:
551 return []
553 if kind_row["kind"] == "condensed":
554 # Two-hop: condensed → leaves (parent_id) → message_events (summary_spans)
555 rows = conn.execute(
556 "SELECT me.id, me.session_id, me.ts, me.content"
557 " FROM message_events me"
558 " JOIN summary_spans ss ON ss.message_event_id = me.id"
559 " JOIN summary_nodes leaf ON leaf.id = ss.summary_id"
560 " WHERE leaf.parent_id = ?"
561 " ORDER BY me.ts, me.id",
562 (summary_id,),
563 ).fetchall()
564 else:
565 # Leaf node (or unknown kind): direct summary_spans traversal
566 rows = conn.execute(
567 "SELECT me.id, me.session_id, me.ts, me.content"
568 " FROM message_events me"
569 " JOIN summary_spans ss ON ss.message_event_id = me.id"
570 " WHERE ss.summary_id = ?"
571 " ORDER BY me.ts, me.id",
572 (summary_id,),
573 ).fetchall()
574 except sqlite3.Error:
575 logger.warning("history_reader: ll_expand query failed", exc_info=True)
576 return []
577 finally:
578 conn.close()
579 return [dict(row) for row in rows]
582def ll_describe(
583 node_id: int,
584 *,
585 db: Path | str = DEFAULT_DB_PATH,
586) -> SummaryNode | None:
587 """Return metadata for a summary_nodes row.
589 Returns ``None`` when the node does not exist or the database is unavailable.
590 """
591 db_path = Path(db)
592 conn = _connect_readonly(db_path)
593 if conn is None:
594 return None
595 try:
596 row = conn.execute(
597 "SELECT id, kind, content, tokens, parent_id, session_id,"
598 " ts_start, ts_end, created_at"
599 " FROM summary_nodes WHERE id = ?",
600 (node_id,),
601 ).fetchone()
602 except sqlite3.Error:
603 logger.warning("history_reader: ll_describe query failed", exc_info=True)
604 return None
605 finally:
606 conn.close()
607 if row is None:
608 return None
609 return SummaryNode(
610 id=row["id"],
611 kind=row["kind"],
612 content=row["content"],
613 tokens=row["tokens"],
614 parent_id=row["parent_id"],
615 session_id=row["session_id"],
616 ts_start=row["ts_start"],
617 ts_end=row["ts_end"],
618 created_at=row["created_at"],
619 )
622# ---------------------------------------------------------------------------
623# Project digest — section providers (ENH-1907)
624# ---------------------------------------------------------------------------
627def _query_touched_files(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
628 try:
629 return conn.execute(
630 "SELECT path, COUNT(*) AS edit_count "
631 "FROM file_events "
632 "WHERE ts >= ? AND path IS NOT NULL "
633 "GROUP BY path "
634 "ORDER BY edit_count DESC, MAX(ts) DESC "
635 "LIMIT ?",
636 (cutoff, cap),
637 ).fetchall()
638 except sqlite3.Error:
639 return []
642def _render_touched_files(rows: list) -> list[str]:
643 lines = []
644 for row in rows:
645 path = row["path"]
646 count = row["edit_count"]
647 noun = "edit" if count == 1 else "edits"
648 lines.append(f"- {path} ({count} {noun})")
649 return lines
652def _query_completed_issues(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
653 try:
654 return conn.execute(
655 "SELECT ts, issue_id, transition, issue_type, priority "
656 "FROM issue_events "
657 "WHERE transition IN ('done', 'cancelled') AND ts >= ? "
658 "ORDER BY ts DESC "
659 "LIMIT ?",
660 (cutoff, cap),
661 ).fetchall()
662 except sqlite3.Error:
663 return []
666def _render_completed_issues(rows: list) -> list[str]:
667 lines = []
668 now = datetime.now(UTC)
669 for row in rows:
670 issue_id = row["issue_id"] or "unknown"
671 ts_str = row["ts"] or ""
672 time_ago = ""
673 if ts_str:
674 try:
675 ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
676 delta_days = (now - ts).days
677 if delta_days == 0:
678 time_ago = " (today)"
679 elif delta_days == 1:
680 time_ago = " (1d ago)"
681 else:
682 time_ago = f" ({delta_days}d ago)"
683 except (ValueError, TypeError):
684 pass
685 lines.append(f"- {issue_id}{time_ago}")
686 return lines
689def _query_recurring_corrections(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list:
690 try:
691 return conn.execute(
692 "SELECT content, COUNT(*) AS seen_count "
693 "FROM user_corrections "
694 "WHERE ts >= ? "
695 "GROUP BY content "
696 "ORDER BY seen_count DESC, MAX(ts) DESC "
697 "LIMIT ?",
698 (cutoff, cap),
699 ).fetchall()
700 except sqlite3.Error:
701 return []
704def _render_recurring_corrections(rows: list) -> list[str]:
705 lines = []
706 for row in rows:
707 content = row["content"] or ""
708 count = row["seen_count"]
709 if len(content) > 80:
710 content = content[:77] + "..."
711 count_str = f" (seen {count}x)" if count > 1 else ""
712 lines.append(f'- "{content}"{count_str}')
713 return lines
716SECTION_PROVIDERS: dict[str, SectionProvider] = {
717 "touched_files": SectionProvider(
718 name="touched_files",
719 query=_query_touched_files,
720 default_cap=10,
721 render=_render_touched_files,
722 ),
723 "completed_issues": SectionProvider(
724 name="completed_issues",
725 query=_query_completed_issues,
726 default_cap=5,
727 render=_render_completed_issues,
728 ),
729 "recurring_corrections": SectionProvider(
730 name="recurring_corrections",
731 query=_query_recurring_corrections,
732 default_cap=5,
733 render=_render_recurring_corrections,
734 ),
735}
737_SECTION_HEADERS: dict[str, str] = {
738 "touched_files": "Recently touched (last {days} days)",
739 "completed_issues": "Recently completed issues",
740 "recurring_corrections": "Recurring corrections",
741}
744def project_digest(
745 db_path: Path,
746 *,
747 days: int = 7,
748 sections: list[str] | None = None,
749) -> ProjectDigest:
750 """Aggregate a project-wide context snapshot from history.db.
752 Returns a :class:`ProjectDigest` with ``.empty == True`` on missing /
753 empty / stale DB. ``sections=None`` or ``sections=[]`` renders all
754 registered providers in registry order; a non-empty list restricts and
755 orders the output.
756 """
757 conn = _connect_readonly(db_path)
758 if conn is None:
759 return ProjectDigest(sections=[], days=days)
761 cutoff = _stale_cutoff(days)
762 provider_keys: list[str] = list(SECTION_PROVIDERS.keys()) if sections is None else sections
764 result: list[tuple[str, list[str]]] = []
765 try:
766 for key in provider_keys:
767 provider = SECTION_PROVIDERS.get(key)
768 if provider is None:
769 logger.warning("project_digest: unknown section %r — skipping", key)
770 continue
771 rows = provider.query(conn, cutoff=cutoff, cap=provider.default_cap)
772 lines = provider.render(rows) if rows else []
773 if lines:
774 result.append((key, lines))
775 finally:
776 conn.close()
778 return ProjectDigest(sections=result, days=days)
781def render_project_context(
782 digest: ProjectDigest,
783 *,
784 char_cap: int = 1200,
785 days: int | None = None,
786) -> str:
787 """Render a ``<project_context>`` block from *digest*, capped at *char_cap* chars.
789 Returns ``""`` when the digest is empty (no block injected). Truncates
790 with a ``+N more`` tail when content would exceed *char_cap*.
791 """
792 if digest.empty:
793 return ""
795 effective_days = days if days is not None else digest.days
796 content_lines: list[str] = []
797 for name, section_lines in digest.sections:
798 raw_header = _SECTION_HEADERS.get(name, name.replace("_", " ").title())
799 header = raw_header.format(days=effective_days)
800 content_lines.append(f"## {header}")
801 content_lines.extend(section_lines)
803 open_tag = "<project_context>"
804 close_tag = "</project_context>"
806 full_block = "\n".join([open_tag] + content_lines + [close_tag])
807 if len(full_block) <= char_cap:
808 return full_block
810 # Truncate: accumulate content lines under budget, append "+N more" tail.
811 close_cost = len("\n" + close_tag)
812 budget = char_cap - len(open_tag) - close_cost - 1 # -1 for opening newline
813 accepted: list[str] = []
814 dropped = 0
815 for line in content_lines:
816 line_cost = len(line) + 1 # +1 for the newline separator
817 if budget >= line_cost:
818 accepted.append(line)
819 budget -= line_cost
820 else:
821 dropped += 1
823 if dropped:
824 tail = f"... +{dropped} more"
825 if budget >= len(tail) + 1:
826 accepted.append(tail)
828 return "\n".join([open_tag] + accepted + [close_tag])