Coverage for little_loops / history_reader.py: 25%

1"""Typed read-only query module for ``.ll/history.db`` (ENH-1752). 

2 

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. 

7 

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 project_digest(db_path, ...) -> ProjectDigest 

32 render_project_context(digest, ...) -> str 

33""" 

34 

35from __future__ import annotations 

36 

37import logging 

38import re 

39import sqlite3 

40from collections.abc import Callable 

41from dataclasses import dataclass 

42from datetime import UTC, datetime, timedelta 

43from pathlib import Path 

44from typing import Any 

45 

46from little_loops.session_store import DEFAULT_DB_PATH, ensure_db 

47 

48logger = logging.getLogger(__name__) 

49 

50STALE_DAYS_DEFAULT = 30 

51 

52 

53# --------------------------------------------------------------------------- 

54# Dataclasses 

55# --------------------------------------------------------------------------- 

56 

57 

58@dataclass 

59class UserCorrection: 

60 ts: str 

61 session_id: str | None 

62 content: str 

63 source: str | None 

64 

65 

66@dataclass 

67class FileEvent: 

68 ts: str 

69 session_id: str | None 

70 path: str | None 

71 op: str | None 

72 issue_id: str | None 

73 git_sha: str | None 

74 

75 

76@dataclass 

77class SearchResult: 

78 content: str 

79 kind: str 

80 ref: str 

81 anchor: str 

82 ts: str 

83 score: float 

84 

85 

86@dataclass 

87class IssueEvent: 

88 ts: str 

89 issue_id: str | None 

90 transition: str | None 

91 discovered_by: str | None 

92 issue_type: str | None 

93 priority: str | None 

94 

95 

96@dataclass 

97class SessionRef: 

98 """A session that co-occurred with an issue's active period (ENH-1711).""" 

99 

100 issue_id: str | None 

101 session_id: str | None 

102 jsonl_path: str | None 

103 first_message_ts: str | None 

104 last_message_ts: str | None 

105 

106 

107@dataclass 

108class SummaryNode: 

109 """A summary_nodes row from the LCM-style compaction DAG (FEAT-1712).""" 

110 

111 id: int 

112 kind: str 

113 content: str 

114 tokens: int | None 

115 parent_id: int | None 

116 session_id: str | None 

117 ts_start: str | None 

118 ts_end: str | None 

119 created_at: str 

120 level: int | None 

121 

122 

123@dataclass 

124class GrepResult: 

125 """A message_event regex match with its covering summary node context (FEAT-1712).""" 

126 

127 message_event_id: int 

128 session_id: str | None 

129 ts: str 

130 content: str 

131 summary_id: int | None 

132 summary_kind: str | None 

133 

134 

135@dataclass(frozen=True) 

136class SectionProvider: 

137 """Config-addressable digest section with query and render logic (ENH-1907).""" 

138 

139 name: str 

140 query: Callable # (conn, *, cutoff: str, cap: int) -> list 

141 default_cap: int 

142 render: Callable # (rows: list) -> list[str] 

143 

144 

145@dataclass 

146class ProjectDigest: 

147 """Aggregated project-context snapshot from history.db (ENH-1907).""" 

148 

149 sections: list[tuple[str, list[str]]] # [(name, markdown_lines), ...] in config order 

150 days: int = 7 

151 

152 @property 

153 def empty(self) -> bool: 

154 return all(not lines for _, lines in self.sections) 

155 

156 

157# --------------------------------------------------------------------------- 

158# Helpers 

159# --------------------------------------------------------------------------- 

160 

161 

162def _stale_cutoff(days: int) -> str: 

163 """ISO-8601 timestamp *days* ago.""" 

164 return (datetime.now(UTC) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ") 

165 

166 

167def _connect_readonly(db_path: Path) -> sqlite3.Connection | None: 

168 """Open a read-only connection, or return None on failure.""" 

169 try: 

170 ensure_db(db_path) 

171 except sqlite3.Error: 

172 logger.warning("history_reader: could not ensure schema for %s", db_path, exc_info=True) 

173 return None 

174 try: 

175 conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) 

176 conn.row_factory = sqlite3.Row 

177 conn.execute("PRAGMA query_only = ON") 

178 except sqlite3.Error: 

179 logger.warning("history_reader: could not open %s read-only", db_path, exc_info=True) 

180 return None 

181 return conn 

182 

183 

184def _row_to_dataclass(row: sqlite3.Row, dc: type[Any]) -> Any: 

185 """Map a sqlite3.Row to a dataclass instance, catching extra/unknown keys.""" 

186 field_names = {f.name for f in dc.__dataclass_fields__.values()} 

187 kwargs = {k: row[k] for k in field_names if k in row.keys()} 

188 return dc(**kwargs) 

189 

190 

191# --------------------------------------------------------------------------- 

192# Query API 

193# --------------------------------------------------------------------------- 

194 

195 

196def find_user_corrections( 

197 topic: str, 

198 *, 

199 limit: int = 10, 

200 include_stale: bool = False, 

201 db: Path | str = DEFAULT_DB_PATH, 

202) -> list[UserCorrection]: 

203 """Return user corrections whose content matches *topic* (LIKE search). 

204 

205 Stale rows (>30 days by default) are excluded unless *include_stale* is set. 

206 """ 

207 db_path = Path(db) 

208 conn = _connect_readonly(db_path) 

209 if conn is None: 

210 return [] 

211 try: 

212 params: list[Any] = [f"%{topic}%"] 

213 where = "WHERE content LIKE ?" 

214 if not include_stale: 

215 where += " AND ts >= ?" 

216 params.append(_stale_cutoff(STALE_DAYS_DEFAULT)) 

217 rows = conn.execute( 

218 f"SELECT ts, session_id, content, source FROM user_corrections {where} " 

219 f"ORDER BY ts DESC LIMIT ?", 

220 (*params, limit), 

221 ).fetchall() 

222 except sqlite3.Error: 

223 logger.warning("history_reader: find_user_corrections query failed", exc_info=True) 

224 return [] 

225 finally: 

226 conn.close() 

227 return [_row_to_dataclass(row, UserCorrection) for row in rows] 

228 

229 

230def recent_file_events( 

231 path: str, 

232 *, 

233 limit: int = 10, 

234 include_stale: bool = False, 

235 db: Path | str = DEFAULT_DB_PATH, 

236) -> list[FileEvent]: 

237 """Return recent file events for *path* (LIKE pattern match). 

238 

239 Stale rows (>30 days by default) are excluded unless *include_stale* is set. 

240 """ 

241 db_path = Path(db) 

242 conn = _connect_readonly(db_path) 

243 if conn is None: 

244 return [] 

245 try: 

246 params: list[Any] = [f"%{path}%"] 

247 where = "WHERE path LIKE ?" 

248 if not include_stale: 

249 where += " AND ts >= ?" 

250 params.append(_stale_cutoff(STALE_DAYS_DEFAULT)) 

251 rows = conn.execute( 

252 f"SELECT ts, session_id, path, op, issue_id, git_sha FROM file_events {where} " 

253 f"ORDER BY ts DESC LIMIT ?", 

254 (*params, limit), 

255 ).fetchall() 

256 except sqlite3.Error: 

257 logger.warning("history_reader: recent_file_events query failed", exc_info=True) 

258 return [] 

259 finally: 

260 conn.close() 

261 return [_row_to_dataclass(row, FileEvent) for row in rows] 

262 

263 

264def search( 

265 query: str, 

266 *, 

267 kind: str | None = None, 

268 limit: int = 10, 

269 db: Path | str = DEFAULT_DB_PATH, 

270) -> list[SearchResult]: 

271 """FTS5 full-text search with optional *kind* filter (tool, file, issue, loop, correction, message). 

272 

273 Returns BM25-ranked results. Gracefully handles invalid FTS5 query syntax. 

274 """ 

275 db_path = Path(db) 

276 conn = _connect_readonly(db_path) 

277 if conn is None: 

278 return [] 

279 try: 

280 if kind: 

281 rows = conn.execute( 

282 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score " 

283 "FROM search_index WHERE search_index MATCH ? AND kind = ? " 

284 "ORDER BY score LIMIT ?", 

285 (query, kind, limit), 

286 ).fetchall() 

287 else: 

288 rows = conn.execute( 

289 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score " 

290 "FROM search_index WHERE search_index MATCH ? " 

291 "ORDER BY score LIMIT ?", 

292 (query, limit), 

293 ).fetchall() 

294 except sqlite3.OperationalError as exc: 

295 logger.warning("history_reader: invalid FTS5 query %r: %s", query, exc) 

296 return [] 

297 finally: 

298 conn.close() 

299 return [_row_to_dataclass(row, SearchResult) for row in rows] 

300 

301 

302def related_issue_events( 

303 issue_id: str, 

304 *, 

305 limit: int = 20, 

306 db: Path | str = DEFAULT_DB_PATH, 

307) -> list[IssueEvent]: 

308 """Return issue events for *issue_id*, ordered by most recent first. 

309 

310 Searches both the ``issue_events`` table and the FTS5 index for cross-references. 

311 """ 

312 db_path = Path(db) 

313 conn = _connect_readonly(db_path) 

314 if conn is None: 

315 return [] 

316 try: 

317 rows = conn.execute( 

318 "SELECT ts, issue_id, transition, discovered_by, issue_type, priority " 

319 "FROM issue_events WHERE issue_id = ? " 

320 "ORDER BY ts DESC LIMIT ?", 

321 (issue_id, limit), 

322 ).fetchall() 

323 except sqlite3.Error: 

324 logger.warning("history_reader: related_issue_events query failed", exc_info=True) 

325 return [] 

326 finally: 

327 conn.close() 

328 return [_row_to_dataclass(row, IssueEvent) for row in rows] 

329 

330 

331def sessions_for_issue( 

332 issue_id: str, 

333 *, 

334 limit: int = 20, 

335 db: Path | str = DEFAULT_DB_PATH, 

336) -> list[SessionRef]: 

337 """Return sessions that co-occurred with *issue_id*'s active period. 

338 

339 Queries the ``issue_sessions`` VIEW (v5 migration, ENH-1711), which joins 

340 ``issue_events`` to ``message_events`` via overlapping timestamps. Live-emitted 

341 rows (from ``issue_lifecycle.py````'s 6 emit sites) now populate ``captured_at`` 

342 directly; no prior ``backfill`` pass is needed for issues processed after 

343 ENH-1839. 

344 

345 Returns an empty list when the view is absent (pre-v5 schema), the issue 

346 has no recorded sessions, or the database is unavailable. 

347 """ 

348 db_path = Path(db) 

349 conn = _connect_readonly(db_path) 

350 if conn is None: 

351 return [] 

352 try: 

353 rows = conn.execute( 

354 "SELECT issue_id, session_id, jsonl_path, first_message_ts, last_message_ts " 

355 "FROM issue_sessions WHERE issue_id = ? " 

356 "ORDER BY first_message_ts DESC LIMIT ?", 

357 (issue_id, limit), 

358 ).fetchall() 

359 except sqlite3.Error: 

360 logger.warning("history_reader: sessions_for_issue query failed", exc_info=True) 

361 return [] 

362 finally: 

363 conn.close() 

364 return [_row_to_dataclass(row, SessionRef) for row in rows] 

365 

366 

367def issue_effort( 

368 issue_id: str, 

369 *, 

370 db: Path | str = DEFAULT_DB_PATH, 

371) -> dict | None: 

372 """Per-issue effort: session_count and cycle_time_days (first→last session). 

373 

374 Returns None when the DB is absent, no sessions exist for the issue, or a 

375 query error occurs. Does NOT reuse sessions_for_issue() to avoid the LIMIT=20 

376 cap which would produce incorrect cycle_time_days for issues with >20 sessions. 

377 """ 

378 db_path = Path(db) 

379 conn = _connect_readonly(db_path) 

380 if conn is None: 

381 return None 

382 try: 

383 row = conn.execute( 

384 "SELECT COUNT(*) AS session_count, MIN(first_message_ts) AS first_ts, " 

385 "MAX(last_message_ts) AS last_ts FROM issue_sessions WHERE issue_id = ?", 

386 (issue_id,), 

387 ).fetchone() 

388 except sqlite3.Error: 

389 logger.warning("history_reader: issue_effort query failed", exc_info=True) 

390 return None 

391 finally: 

392 conn.close() 

393 if row is None or row["session_count"] == 0: 

394 return None 

395 cycle: float | None = None 

396 if row["first_ts"] and row["last_ts"]: 

397 delta = datetime.fromisoformat(row["last_ts"]) - datetime.fromisoformat(row["first_ts"]) 

398 cycle = delta.total_seconds() / 86400 

399 return {"session_count": row["session_count"], "cycle_time_days": cycle} 

400 

401 

402def recent_issue_velocity( 

403 limit: int = 10, 

404 *, 

405 db: Path | str = DEFAULT_DB_PATH, 

406) -> list[dict]: 

407 """Effort data for recently completed issues; empty list when DB has no data. 

408 

409 Queries issue_events for recently-completed issues (non-NULL completed_at), 

410 then calls issue_effort() for each to produce per-issue effort dicts. 

411 """ 

412 db_path = Path(db) 

413 conn = _connect_readonly(db_path) 

414 if conn is None: 

415 return [] 

416 try: 

417 rows = conn.execute( 

418 "SELECT DISTINCT issue_id FROM issue_events " 

419 "WHERE completed_at IS NOT NULL " 

420 "ORDER BY completed_at DESC LIMIT ?", 

421 (limit,), 

422 ).fetchall() 

423 except sqlite3.Error: 

424 logger.warning("history_reader: recent_issue_velocity query failed", exc_info=True) 

425 return [] 

426 finally: 

427 conn.close() 

428 result = [] 

429 for row in rows: 

430 effort = issue_effort(row["issue_id"], db=db) 

431 if effort is not None: 

432 result.append({"issue_id": row["issue_id"], **effort}) 

433 return result 

434 

435 

436def lookup_session_metadata( 

437 session_id: str, 

438 *, 

439 db: Path | str = DEFAULT_DB_PATH, 

440) -> dict: 

441 """Return session-quality metadata dict for a session ID (ENH-1943). 

442 

443 Returns: 

444 dict with keys: ``has_corrections`` (bool), ``issue_outcome`` (str|None), 

445 ``tool_count`` (int), ``files_modified`` (int), ``loop_outcome`` (str|None). 

446 

447 ``loop_outcome`` is always ``None`` until ``loop_events`` gains a 

448 ``session_id`` column (schema change out of scope). 

449 

450 Returns empty dict ``{}`` when DB is missing, empty, or lacks relevant tables. 

451 """ 

452 db_path = Path(db) 

453 if not db_path.exists(): 

454 return {} 

455 conn = _connect_readonly(db_path) 

456 if conn is None: 

457 return {} 

458 try: 

459 # has_corrections: direct query on user_corrections 

460 row = conn.execute( 

461 "SELECT COUNT(*) > 0 AS has_corrections FROM user_corrections WHERE session_id = ?", 

462 (session_id,), 

463 ).fetchone() 

464 has_corrections = bool(row["has_corrections"]) if row else False 

465 

466 # issue_outcome: JOIN through issue_sessions VIEW (issue_events has no 

467 # session_id column; migration v5 bridges via the VIEW) 

468 row = conn.execute( 

469 "SELECT ie.transition " 

470 "FROM issue_sessions is2 " 

471 "JOIN issue_events ie ON is2.issue_id = ie.issue_id " 

472 "WHERE is2.session_id = ? AND ie.transition = 'done' " 

473 "ORDER BY ie.ts DESC LIMIT 1", 

474 (session_id,), 

475 ).fetchone() 

476 issue_outcome: str | None = row["transition"] if row else None 

477 

478 # tool_count: direct query on tool_events 

479 row = conn.execute( 

480 "SELECT COUNT(*) AS tool_count FROM tool_events WHERE session_id = ?", 

481 (session_id,), 

482 ).fetchone() 

483 tool_count: int = row["tool_count"] if row else 0 

484 

485 # files_modified: direct query on file_events; op values include both 

486 # hook-written tool names ('Write') and lowercase variants ('write', 'create') 

487 row = conn.execute( 

488 "SELECT COUNT(*) AS files_modified FROM file_events " 

489 "WHERE session_id = ? AND op IN ('write', 'create', 'Write')", 

490 (session_id,), 

491 ).fetchone() 

492 files_modified: int = row["files_modified"] if row else 0 

493 

494 # loop_outcome: loop_events has no session_id column; always None 

495 loop_outcome: None = None 

496 

497 except sqlite3.Error: 

498 logger.warning("history_reader: lookup_session_metadata query failed", exc_info=True) 

499 return {} 

500 finally: 

501 conn.close() 

502 

503 return { 

504 "has_corrections": has_corrections, 

505 "issue_outcome": issue_outcome, 

506 "tool_count": tool_count, 

507 "files_modified": files_modified, 

508 "loop_outcome": loop_outcome, 

509 } 

510 

511 

512def conversation_turns( 

513 db_path: Path | str, 

514 since: datetime | None = None, 

515 context_window: int = 3, 

516) -> list[list[tuple[str, str]]]: 

517 """Return conversation turn-pair windows from ``history.db`` (ENH-1942). 

518 

519 Queries ``message_events`` and ``assistant_messages``, pairs user messages 

520 with their assistant responses via temporal adjacency (same algorithm as 

521 ``_extract_turn_pairs()`` in ``user_messages.py``), and groups them into 

522 sliding windows of *context_window* turn-pairs each. 

523 

524 Returns ``[]`` when the database is missing, empty, predates schema v11 

525 (no ``assistant_messages`` table), or when no turn-pairs match the *since* 

526 filter. Callers should fall back to JSONL parsing in that case. 

527 

528 Args: 

529 db_path: Path to ``history.db``. 

530 since: Only include turns where the user message timestamp is >= this value. 

531 context_window: Number of (user, assistant) turn-pairs per output window. 

532 

533 Returns: 

534 List of conversation windows; each window is a ``list[tuple[str, str]]`` 

535 alternating between ``("user", text)`` and ``("assistant", text)``. 

536 """ 

537 db_path = Path(db_path) 

538 if not db_path.exists(): 

539 return [] 

540 conn = _connect_readonly(db_path) 

541 if conn is None: 

542 return [] 

543 try: 

544 # Check that assistant_messages table exists (schema >= v11) 

545 row = conn.execute( 

546 "SELECT COUNT(*) AS n FROM sqlite_master " 

547 "WHERE type = 'table' AND name = 'assistant_messages'" 

548 ).fetchone() 

549 if not row or row["n"] == 0: 

550 return [] 

551 

552 # Pair each user message with assistant messages between it and the 

553 # NEXT user message (temporal-adjacency, matching _extract_turn_pairs). 

554 # The subquery finds the next user message timestamp per session; 

555 # COALESCE defaults to a far-future sentinel for the last message. 

556 base_sql = ( 

557 "SELECT u.session_id, u.ts AS user_ts, u.content AS user_text, " 

558 "a.content AS assistant_text " 

559 "FROM message_events u " 

560 "JOIN assistant_messages a ON a.session_id = u.session_id " 

561 "AND a.ts > u.ts " 

562 "AND a.ts < COALESCE(" 

563 " (SELECT MIN(u2.ts) FROM message_events u2 " 

564 " WHERE u2.session_id = u.session_id AND u2.ts > u.ts), " 

565 " '9999-12-31'" 

566 ") " 

567 ) 

568 params: list[Any] = [] 

569 if since is not None: 

570 base_sql += "WHERE u.ts >= ? " 

571 params.append(since.strftime("%Y-%m-%dT%H:%M:%SZ")) 

572 base_sql += "ORDER BY u.session_id, u.ts, a.ts" 

573 

574 rows = conn.execute(base_sql, params or ()).fetchall() 

575 

576 if not rows: 

577 return [] 

578 

579 # Group assistant texts by user message. 

580 # The SQL already only includes assistant messages between consecutive 

581 # user messages, so simple grouping by (session_id, user_ts) suffices. 

582 turn_pairs: list[tuple[str, str]] = [] 

583 current_key: tuple[str, str] | None = None 

584 current_user: str = "" 

585 assistant_texts: list[str] = [] 

586 

587 for row_ in rows: 

588 key = (row_["session_id"], row_["user_ts"]) 

589 if key != current_key: 

590 if current_key is not None and assistant_texts: 

591 turn_pairs.append((current_user, "\n\n".join(assistant_texts))) 

592 current_key = key 

593 current_user = row_["user_text"] 

594 assistant_texts = [] 

595 assistant_texts.append(row_["assistant_text"]) 

596 

597 # Flush the final turn 

598 if current_key is not None and assistant_texts: 

599 turn_pairs.append((current_user, "\n\n".join(assistant_texts))) 

600 

601 # Emit sliding windows of context_window turn-pairs 

602 windows: list[list[tuple[str, str]]] = [] 

603 n = len(turn_pairs) 

604 if n == 0: 

605 return [] 

606 for i in range(max(1, n - context_window + 1)): 

607 window_pairs = turn_pairs[i : i + context_window] 

608 window: list[tuple[str, str]] = [] 

609 for user_text, assistant_text in window_pairs: 

610 window.append(("user", user_text)) 

611 window.append(("assistant", assistant_text)) 

612 windows.append(window) 

613 

614 return windows 

615 

616 except sqlite3.Error: 

617 logger.warning("history_reader: conversation_turns query failed", exc_info=True) 

618 return [] 

619 finally: 

620 conn.close() 

621 

622 

623# --------------------------------------------------------------------------- 

624# Summary DAG retrieval (FEAT-1712) 

625# --------------------------------------------------------------------------- 

626 

627 

628def ll_grep( 

629 pattern: str, 

630 *, 

631 summary_id: int | None = None, 

632 limit: int = 50, 

633 db: Path | str = DEFAULT_DB_PATH, 

634) -> list[GrepResult]: 

635 """Regex search over message_events, with covering summary node context. 

636 

637 Each result includes the summary_id and summary_kind of the leaf node covering the 

638 matched message (or None/None for messages not yet compacted). If *summary_id* is 

639 provided, restrict the search to messages covered by that specific node. 

640 

641 When *summary_id* is a condensed node (kind='condensed'), uses a recursive CTE to 

642 walk the N-level DAG (condensed → … → leaves via parent_id → message_events via 

643 summary_spans) so that messages under all descendant leaves are searched regardless 

644 of condensation depth. 

645 """ 

646 db_path = Path(db) 

647 conn = _connect_readonly(db_path) 

648 if conn is None: 

649 return [] 

650 

651 def _regexp(pat: str, val: str | None) -> bool: 

652 try: 

653 return bool(re.search(pat, val or "", re.IGNORECASE)) 

654 except re.error: 

655 return False 

656 

657 try: 

658 conn.create_function("regexp_match", 2, _regexp) 

659 if summary_id is not None: 

660 # Recursive CTE walks the full N-level DAG from the starting node 

661 # through all descendants, terminating at leaf nodes that have 

662 # summary_spans entries. Works uniformly for both kind='leaf' 

663 # (CTE = 1 row) and kind='condensed' at any depth. 

664 rows = conn.execute( 

665 "WITH RECURSIVE descendants AS (" 

666 " SELECT id, kind FROM summary_nodes WHERE id = ?1" 

667 " UNION ALL" 

668 " SELECT sn.id, sn.kind" 

669 " FROM summary_nodes sn" 

670 " JOIN descendants d ON sn.parent_id = d.id" 

671 ")" 

672 "SELECT me.id, me.session_id, me.ts, me.content," 

673 " sn.id AS summary_id, sn.kind AS summary_kind" 

674 " FROM message_events me" 

675 " JOIN summary_spans ss ON ss.message_event_id = me.id" 

676 " JOIN descendants leaf ON leaf.id = ss.summary_id" 

677 " JOIN summary_nodes sn ON sn.id = leaf.id" 

678 " WHERE regexp_match(?2, me.content)" 

679 " ORDER BY me.ts, me.id LIMIT ?3", 

680 (summary_id, pattern, limit), 

681 ).fetchall() 

682 else: 

683 rows = conn.execute( 

684 "SELECT me.id, me.session_id, me.ts, me.content," 

685 " sn.id AS summary_id, sn.kind AS summary_kind" 

686 " FROM message_events me" 

687 " LEFT JOIN summary_spans ss ON ss.message_event_id = me.id" 

688 " LEFT JOIN summary_nodes sn ON sn.id = ss.summary_id" 

689 " WHERE regexp_match(?, me.content)" 

690 " ORDER BY me.ts, me.id LIMIT ?", 

691 (pattern, limit), 

692 ).fetchall() 

693 except sqlite3.Error: 

694 logger.warning("history_reader: ll_grep query failed", exc_info=True) 

695 return [] 

696 finally: 

697 conn.close() 

698 

699 return [ 

700 GrepResult( 

701 message_event_id=row["id"], 

702 session_id=row["session_id"], 

703 ts=row["ts"], 

704 content=row["content"] or "", 

705 summary_id=row["summary_id"], 

706 summary_kind=row["summary_kind"], 

707 ) 

708 for row in rows 

709 ] 

710 

711 

712def ll_expand( 

713 summary_id: int, 

714 *, 

715 db: Path | str = DEFAULT_DB_PATH, 

716) -> list[dict]: 

717 """Return the message_events covered by *summary_id*. 

718 

719 Uses a recursive CTE to walk the N-level summary DAG from the starting 

720 node through all descendants, terminating at leaf nodes that have 

721 ``summary_spans`` entries. Works uniformly for both ``kind='leaf'`` 

722 (CTE = 1 row, direct span join) and ``kind='condensed'`` at any 

723 condensation depth (CTE descends through intermediate condensed nodes 

724 to reach the leaves). 

725 

726 Returns dicts with keys ``id``, ``session_id``, ``ts``, ``content``. 

727 Empty list when the summary node does not exist or has no spans. 

728 """ 

729 db_path = Path(db) 

730 conn = _connect_readonly(db_path) 

731 if conn is None: 

732 return [] 

733 try: 

734 # Recursive CTE handles leaf and condensed nodes uniformly at any depth. 

735 # For a leaf node, descendants = {itself} and the summary_spans join is direct. 

736 # For a condensed node, descendants = {condensed, children, grandchildren, …} 

737 # and the summary_spans join only matches the leaf-descendant rows. 

738 rows = conn.execute( 

739 "WITH RECURSIVE descendants AS (" 

740 " SELECT id, kind FROM summary_nodes WHERE id = ?1" 

741 " UNION ALL" 

742 " SELECT sn.id, sn.kind" 

743 " FROM summary_nodes sn" 

744 " JOIN descendants d ON sn.parent_id = d.id" 

745 ")" 

746 "SELECT me.id, me.session_id, me.ts, me.content" 

747 " FROM message_events me" 

748 " JOIN summary_spans ss ON ss.message_event_id = me.id" 

749 " JOIN descendants leaf ON leaf.id = ss.summary_id" 

750 " ORDER BY me.ts, me.id", 

751 (summary_id,), 

752 ).fetchall() 

753 except sqlite3.Error: 

754 logger.warning("history_reader: ll_expand query failed", exc_info=True) 

755 return [] 

756 finally: 

757 conn.close() 

758 return [dict(row) for row in rows] 

759 

760 

761def ll_describe( 

762 node_id: int, 

763 *, 

764 db: Path | str = DEFAULT_DB_PATH, 

765) -> SummaryNode | None: 

766 """Return metadata for a summary_nodes row. 

767 

768 Returns ``None`` when the node does not exist or the database is unavailable. 

769 """ 

770 db_path = Path(db) 

771 conn = _connect_readonly(db_path) 

772 if conn is None: 

773 return None 

774 try: 

775 row = conn.execute( 

776 "SELECT id, kind, content, tokens, parent_id, session_id," 

777 " ts_start, ts_end, created_at, level" 

778 " FROM summary_nodes WHERE id = ?", 

779 (node_id,), 

780 ).fetchone() 

781 except sqlite3.Error: 

782 logger.warning("history_reader: ll_describe query failed", exc_info=True) 

783 return None 

784 finally: 

785 conn.close() 

786 if row is None: 

787 return None 

788 return SummaryNode( 

789 id=row["id"], 

790 kind=row["kind"], 

791 content=row["content"], 

792 tokens=row["tokens"], 

793 parent_id=row["parent_id"], 

794 session_id=row["session_id"], 

795 ts_start=row["ts_start"], 

796 ts_end=row["ts_end"], 

797 created_at=row["created_at"], 

798 level=row["level"], 

799 ) 

800 

801 

802# --------------------------------------------------------------------------- 

803# Project digest — section providers (ENH-1907) 

804# --------------------------------------------------------------------------- 

805 

806 

807def _query_touched_files(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list: 

808 try: 

809 return conn.execute( 

810 "SELECT path, COUNT(*) AS edit_count " 

811 "FROM file_events " 

812 "WHERE ts >= ? AND path IS NOT NULL " 

813 "GROUP BY path " 

814 "ORDER BY edit_count DESC, MAX(ts) DESC " 

815 "LIMIT ?", 

816 (cutoff, cap), 

817 ).fetchall() 

818 except sqlite3.Error: 

819 return [] 

820 

821 

822def _render_touched_files(rows: list) -> list[str]: 

823 lines = [] 

824 for row in rows: 

825 path = row["path"] 

826 count = row["edit_count"] 

827 noun = "edit" if count == 1 else "edits" 

828 lines.append(f"- {path} ({count} {noun})") 

829 return lines 

830 

831 

832def _query_completed_issues(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list: 

833 try: 

834 return conn.execute( 

835 "SELECT ts, issue_id, transition, issue_type, priority " 

836 "FROM issue_events " 

837 "WHERE transition IN ('done', 'cancelled') AND ts >= ? " 

838 "ORDER BY ts DESC " 

839 "LIMIT ?", 

840 (cutoff, cap), 

841 ).fetchall() 

842 except sqlite3.Error: 

843 return [] 

844 

845 

846def _render_completed_issues(rows: list) -> list[str]: 

847 lines = [] 

848 now = datetime.now(UTC) 

849 for row in rows: 

850 issue_id = row["issue_id"] or "unknown" 

851 ts_str = row["ts"] or "" 

852 time_ago = "" 

853 if ts_str: 

854 try: 

855 ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) 

856 delta_days = (now - ts).days 

857 if delta_days == 0: 

858 time_ago = " (today)" 

859 elif delta_days == 1: 

860 time_ago = " (1d ago)" 

861 else: 

862 time_ago = f" ({delta_days}d ago)" 

863 except (ValueError, TypeError): 

864 pass 

865 lines.append(f"- {issue_id}{time_ago}") 

866 return lines 

867 

868 

869def _query_recurring_corrections(conn: sqlite3.Connection, *, cutoff: str, cap: int) -> list: 

870 try: 

871 return conn.execute( 

872 "SELECT content, COUNT(*) AS seen_count " 

873 "FROM user_corrections " 

874 "WHERE ts >= ? " 

875 "GROUP BY content " 

876 "ORDER BY seen_count DESC, MAX(ts) DESC " 

877 "LIMIT ?", 

878 (cutoff, cap), 

879 ).fetchall() 

880 except sqlite3.Error: 

881 return [] 

882 

883 

884def _render_recurring_corrections(rows: list) -> list[str]: 

885 lines = [] 

886 for row in rows: 

887 content = row["content"] or "" 

888 count = row["seen_count"] 

889 if len(content) > 80: 

890 content = content[:77] + "..." 

891 count_str = f" (seen {count}x)" if count > 1 else "" 

892 lines.append(f'- "{content}"{count_str}') 

893 return lines 

894 

895 

896SECTION_PROVIDERS: dict[str, SectionProvider] = { 

897 "touched_files": SectionProvider( 

898 name="touched_files", 

899 query=_query_touched_files, 

900 default_cap=10, 

901 render=_render_touched_files, 

902 ), 

903 "completed_issues": SectionProvider( 

904 name="completed_issues", 

905 query=_query_completed_issues, 

906 default_cap=5, 

907 render=_render_completed_issues, 

908 ), 

909 "recurring_corrections": SectionProvider( 

910 name="recurring_corrections", 

911 query=_query_recurring_corrections, 

912 default_cap=5, 

913 render=_render_recurring_corrections, 

914 ), 

915} 

916 

917_SECTION_HEADERS: dict[str, str] = {