Coverage for little_loops / session_store.py: 15%
271 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-28 13:07 -0500
1"""Unified session store: a per-project SQLite + FTS5 database (FEAT-1112).
3A single ``.ll/history.db`` is the per-project event history across all
4Claude Code sessions: it indexes tool events, file modifications, issue
5transitions, loop runs, and user corrections so cross-cutting queries
6("which loops failed on issues touching file X?") can be answered in
7milliseconds rather than re-parsing scattered JSON/markdown sources. The
8``session_id`` column ties each row back to its originating session JSONL,
9but the database itself is long-lived and never rotated.
11The store is purely additive: it never replaces an existing data path. The
12``SQLiteTransport`` sink subscribes to the EventBus alongside the other
13transports, and the backfill routine seeds the database from on-disk sources
14that the analyze-* skills already read.
16Public API:
17 DEFAULT_DB_PATH: default database location (``.ll/history.db``)
18 SCHEMA_VERSION: current schema version integer
19 ensure_db(path): create the database and apply pending migrations
20 connect(path): open a connection (ensures schema first)
21 SQLiteTransport: EventBus Transport sink writing FSM events to
22 ``loop_events`` and issue lifecycle events to
23 ``issue_events`` (ENH-1690)
24 backfill(db,...): populate the database from existing on-disk sources
25 search(db,...): FTS5 full-text query with BM25 ranking
26 recent(db,...): recent rows for a given event kind
27"""
29from __future__ import annotations
31import hashlib
32import json
33import logging
34import re
35import sqlite3
36import threading
37from datetime import UTC, datetime
38from pathlib import Path
39from typing import Any
41logger = logging.getLogger(__name__)
43DEFAULT_DB_PATH = Path(".ll/history.db")
44SCHEMA_VERSION = 3
46_VALID_KINDS = frozenset({"tool", "file", "issue", "loop", "correction", "message"})
47_KIND_TABLE = {
48 "tool": "tool_events",
49 "file": "file_events",
50 "issue": "issue_events",
51 "loop": "loop_events",
52 "correction": "user_corrections",
53 "message": "message_events",
54}
56# FSM event types the SQLiteTransport records as loop_events rows.
57_LOOP_EVENT_TYPES = frozenset(
58 {
59 "loop_start",
60 "loop_resume",
61 "loop_complete",
62 "state_enter",
63 "route",
64 "retry_exhausted",
65 "cycle_detected",
66 "max_iterations_summary",
67 }
68)
71# ---------------------------------------------------------------------------
72# Schema + migrations
73# ---------------------------------------------------------------------------
75# Each entry is the full SQL applied to move the schema from version index to
76# index+1. Migration 0 bootstraps the whole schema; append new entries to
77# evolve it. ``bytes_in`` / ``bytes_out`` / ``cache_hit`` are reserved on
78# ``tool_events`` for FEAT-1160 (Context Window Analytics) so that feature does
79# not require a follow-up migration.
80_MIGRATIONS: list[str] = [
81 """
82 CREATE TABLE tool_events (
83 id INTEGER PRIMARY KEY AUTOINCREMENT,
84 ts TEXT NOT NULL,
85 session_id TEXT,
86 tool_name TEXT,
87 args_hash TEXT,
88 result_size INTEGER,
89 bytes_in INTEGER,
90 bytes_out INTEGER,
91 cache_hit INTEGER
92 );
93 CREATE TABLE file_events (
94 id INTEGER PRIMARY KEY AUTOINCREMENT,
95 ts TEXT NOT NULL,
96 session_id TEXT,
97 path TEXT,
98 op TEXT,
99 issue_id TEXT,
100 git_sha TEXT
101 );
102 CREATE TABLE issue_events (
103 id INTEGER PRIMARY KEY AUTOINCREMENT,
104 ts TEXT NOT NULL,
105 issue_id TEXT,
106 transition TEXT,
107 discovered_by TEXT
108 );
109 CREATE TABLE loop_events (
110 id INTEGER PRIMARY KEY AUTOINCREMENT,
111 ts TEXT NOT NULL,
112 loop_name TEXT,
113 state TEXT,
114 transition TEXT,
115 retries INTEGER
116 );
117 CREATE TABLE user_corrections (
118 id INTEGER PRIMARY KEY AUTOINCREMENT,
119 ts TEXT NOT NULL,
120 session_id TEXT,
121 content TEXT,
122 source TEXT
123 );
124 CREATE VIRTUAL TABLE search_index USING fts5(
125 content,
126 kind UNINDEXED,
127 ref UNINDEXED,
128 anchor UNINDEXED,
129 ts UNINDEXED
130 );
131 CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT);
132 """,
133 # v2 (ENH-1621): widen issue_events with completion-summary columns so
134 # ll-history `summary` can be answered from the DB; add message_events for
135 # analyze_workflows() to read user message bodies without re-parsing JSONL.
136 """
137 ALTER TABLE issue_events ADD COLUMN issue_type TEXT;
138 ALTER TABLE issue_events ADD COLUMN priority TEXT;
139 ALTER TABLE issue_events ADD COLUMN completed_date TEXT;
140 ALTER TABLE issue_events ADD COLUMN captured_at TEXT;
141 ALTER TABLE issue_events ADD COLUMN completed_at TEXT;
142 CREATE TABLE message_events (
143 id INTEGER PRIMARY KEY AUTOINCREMENT,
144 ts TEXT NOT NULL,
145 session_id TEXT,
146 content TEXT
147 );
148 """,
149 # v3 (ENH-1690): unique dedup index on issue_events so INSERT OR IGNORE
150 # prevents duplicate rows when backfill() is called multiple times.
151 """
152 CREATE UNIQUE INDEX IF NOT EXISTS idx_issue_events_dedup
153 ON issue_events(issue_id, transition);
154 """,
155]
158def _now() -> str:
159 """Return the current UTC time as a Z-suffixed ISO 8601 string."""
160 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
163def _current_version(conn: sqlite3.Connection) -> int:
164 """Return the applied schema version, or 0 if the meta table is absent."""
165 try:
166 row = conn.execute("SELECT value FROM meta WHERE key = 'schema_version'").fetchone()
167 except sqlite3.OperationalError:
168 return 0
169 return int(row[0]) if row else 0
172def _apply_migrations(conn: sqlite3.Connection) -> None:
173 """Apply every migration newer than the database's current version."""
174 version = _current_version(conn)
175 for index in range(version, len(_MIGRATIONS)):
176 conn.executescript(_MIGRATIONS[index])
177 conn.execute(
178 "INSERT INTO meta(key, value) VALUES('schema_version', ?) "
179 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
180 (str(index + 1),),
181 )
182 conn.commit()
185def ensure_db(path: Path | str = DEFAULT_DB_PATH) -> Path:
186 """Create the database at *path* (if needed) and apply pending migrations.
188 Idempotent: safe to call on every session start. The parent directory is
189 created if absent. Returns the resolved database path.
191 On the first call after the ENH-1635 rename, transparently migrates a
192 pre-existing ``.ll/session.db`` (and any ``-shm``/``-wal`` sidecars) to
193 the new ``.ll/history.db`` path. Each sidecar is renamed independently so
194 a single failure does not abort the others; failures are logged at
195 WARNING (the caller in ``hooks/session_start.py`` wraps the whole call
196 in ``contextlib.suppress(Exception)``, which would otherwise silence
197 diagnostic context).
198 """
199 db_path = Path(path)
200 legacy = db_path.parent / "session.db"
201 if legacy.exists() and not db_path.exists():
202 for suffix in ("", "-shm", "-wal"):
203 src = legacy.parent / f"session.db{suffix}"
204 if src.exists():
205 dst = db_path.parent / f"history.db{suffix}"
206 try:
207 src.rename(dst)
208 logger.info("session_store: migrated %s -> %s", src, dst)
209 except OSError:
210 logger.warning(
211 "session_store: legacy rename failed for %s; continuing with fresh db",
212 src,
213 exc_info=True,
214 )
215 break
216 db_path.parent.mkdir(parents=True, exist_ok=True)
217 conn = sqlite3.connect(str(db_path))
218 try:
219 _apply_migrations(conn)
220 finally:
221 conn.close()
222 return db_path
225def connect(path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
226 """Open a connection to the session database, ensuring the schema first.
228 Rows are returned as :class:`sqlite3.Row` so callers can index by name.
229 """
230 db_path = ensure_db(path)
231 conn = sqlite3.connect(str(db_path))
232 conn.row_factory = sqlite3.Row
233 return conn
236def _index(
237 conn: sqlite3.Connection,
238 *,
239 content: str,
240 kind: str,
241 ref: str,
242 anchor: str,
243 ts: str,
244) -> None:
245 """Insert one row into the FTS5 ``search_index`` table."""
246 conn.execute(
247 "INSERT INTO search_index(content, kind, ref, anchor, ts) VALUES(?, ?, ?, ?, ?)",
248 (content, kind, ref, anchor, ts),
249 )
252# ---------------------------------------------------------------------------
253# Query API
254# ---------------------------------------------------------------------------
257def search(
258 db: Path | str = DEFAULT_DB_PATH,
259 *,
260 query: str,
261 limit: int = 20,
262) -> list[dict[str, Any]]:
263 """Run an FTS5 full-text query, returning BM25-ranked results.
265 Each result dict carries ``content``, ``kind``, ``ref``, ``anchor`` (a
266 file:line-style pointer where available), ``ts`` and a numeric ``score``
267 (lower BM25 score = better match).
268 """
269 conn = connect(db)
270 try:
271 rows = conn.execute(
272 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
273 "FROM search_index WHERE search_index MATCH ? "
274 "ORDER BY score LIMIT ?",
275 (query, limit),
276 ).fetchall()
277 except sqlite3.OperationalError as exc:
278 raise ValueError(f"invalid FTS query {query!r}: {exc}") from exc
279 finally:
280 conn.close()
281 return [dict(row) for row in rows]
284def recent(
285 db: Path | str = DEFAULT_DB_PATH,
286 *,
287 kind: str,
288 limit: int = 20,
289) -> list[dict[str, Any]]:
290 """Return the most recent rows for *kind* (tool, file, issue, loop, correction)."""
291 if kind not in _VALID_KINDS:
292 raise ValueError(f"unknown kind {kind!r}; expected one of {sorted(_VALID_KINDS)}")
293 table = _KIND_TABLE[kind]
294 conn = connect(db)
295 try:
296 rows = conn.execute(
297 f"SELECT * FROM {table} ORDER BY id DESC LIMIT ?", # noqa: S608 - table from fixed map
298 (limit,),
299 ).fetchall()
300 finally:
301 conn.close()
302 return [dict(row) for row in rows]
305# ---------------------------------------------------------------------------
306# SQLiteTransport
307# ---------------------------------------------------------------------------
309_ISSUE_TRANSITION_MAP: dict[str, str] = {
310 "issue.completed": "done",
311 "issue.closed": "done",
312 "issue.deferred": "deferred",
313 "issue.skipped": "cancelled",
314 "issue.created": "open",
315 "issue.started": "in_progress",
316}
319def _derive_transition(event_type: str) -> str:
320 """Map an ``issue.*`` event type to the canonical transition/status string."""
321 return _ISSUE_TRANSITION_MAP.get(event_type, event_type.split(".", 1)[1])
324class SQLiteTransport:
325 """EventBus sink that records FSM loop events into the session database.
327 A single connection is opened at construction with ``check_same_thread``
328 disabled, since :meth:`send` may be called from the FSM thread while other
329 transports run their own threads; a lock serialises writes. Every
330 operation is best-effort — a database error is logged and swallowed so a
331 failing sink never aborts a loop run (the four ``wire_transports`` call
332 sites depend on this).
333 """
335 def __init__(self, db_path: Path | str = DEFAULT_DB_PATH) -> None:
336 self._path = Path(db_path)
337 self._lock = threading.Lock()
338 self._conn: sqlite3.Connection | None = None
339 try:
340 ensure_db(self._path)
341 self._conn = sqlite3.connect(str(self._path), check_same_thread=False)
342 except sqlite3.Error:
343 logger.warning(
344 "SQLiteTransport: could not open %s; sink disabled", self._path, exc_info=True
345 )
346 self._conn = None
348 def send(self, event: dict[str, Any]) -> None:
349 """Record a recognised event as a ``loop_events`` or ``issue_events`` row (best-effort)."""
350 conn = self._conn
351 if conn is None:
352 return
353 event_type = str(event.get("event", ""))
354 ts = str(event.get("ts") or _now())
355 try:
356 with self._lock:
357 if event_type in _LOOP_EVENT_TYPES:
358 loop_name = str(event.get("loop_name", "")) or None
359 state = event.get("state")
360 if event_type == "loop_complete":
361 state = event.get("outcome", state)
362 retries = event.get("retries")
363 conn.execute(
364 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
365 "VALUES(?, ?, ?, ?, ?)",
366 (
367 ts,
368 loop_name,
369 str(state) if state is not None else None,
370 event_type,
371 int(retries) if isinstance(retries, int) else None,
372 ),
373 )
374 _index(
375 conn,
376 content=" ".join(
377 str(p) for p in (loop_name, state, event_type) if p is not None
378 ),
379 kind="loop",
380 ref=loop_name or "",
381 anchor=f".loops/{loop_name}.yaml" if loop_name else "",
382 ts=ts,
383 )
384 elif event_type.startswith("issue."):
385 issue_id = event.get("issue_id")
386 transition = _derive_transition(event_type)
387 conn.execute(
388 "INSERT OR IGNORE INTO issue_events("
389 "ts, issue_id, transition, discovered_by, "
390 "issue_type, priority, captured_at, completed_at"
391 ") VALUES(?,?,?,?,?,?,?,?)",
392 (
393 ts,
394 issue_id,
395 transition,
396 event.get("discovered_by"),
397 event.get("issue_type"),
398 event.get("priority"),
399 event.get("captured_at"),
400 event.get("completed_at"),
401 ),
402 )
403 _index(
404 conn,
405 content=f"{issue_id or ''} {event.get('issue_type', '')}".strip(),
406 kind="issue",
407 ref=str(issue_id or ""),
408 anchor=event.get("issue_file", ""),
409 ts=ts,
410 )
411 else:
412 return
413 conn.commit()
414 except sqlite3.Error:
415 logger.warning("SQLiteTransport: write failed for event %r", event_type, exc_info=True)
417 def close(self) -> None:
418 """Close the underlying connection (best-effort)."""
419 if self._conn is not None:
420 try:
421 self._conn.close()
422 except sqlite3.Error:
423 pass
424 self._conn = None
427# ---------------------------------------------------------------------------
428# Backfill
429# ---------------------------------------------------------------------------
432def _hash_args(value: Any) -> str:
433 """Return a short stable hash of a tool-call argument structure."""
434 try:
435 blob = json.dumps(value, sort_keys=True, default=str)
436 except (TypeError, ValueError):
437 blob = repr(value)
438 return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
441_FILENAME_TYPE_RE = re.compile(r"(BUG|ENH|FEAT|EPIC)-(\d+)")
442_FILENAME_PRIORITY_RE = re.compile(r"^(P\d)")
445def _derive_type_priority(filename: str, fm: dict[str, Any]) -> tuple[str | None, str | None]:
446 """Derive (issue_type, priority) preferring frontmatter, falling back to filename.
448 Mirrors :func:`little_loops.issue_history.parsing.parse_completed_issue`'s
449 filename-parsing convention (``P[0-5]-[TYPE]-[NNN]-...``).
450 """
451 fm_type = fm.get("type")
452 issue_type: str | None = str(fm_type) if isinstance(fm_type, str) and fm_type else None
453 fm_priority = fm.get("priority")
454 priority: str | None = (
455 str(fm_priority) if isinstance(fm_priority, str) and fm_priority else None
456 )
457 if issue_type is None:
458 m = _FILENAME_TYPE_RE.search(filename)
459 if m:
460 issue_type = m.group(1)
461 if priority is None:
462 m = _FILENAME_PRIORITY_RE.match(filename)
463 if m:
464 priority = m.group(1)
465 return issue_type, priority
468def _backfill_issues(conn: sqlite3.Connection, issues_dir: Path) -> int:
469 """Seed ``issue_events`` from issue-file frontmatter under *issues_dir*.
471 Populates the v2 summary columns (``issue_type``, ``priority``,
472 ``completed_date``, ``captured_at``, ``completed_at``) so ``ll-history
473 summary`` can be answered from the DB without re-reading the files
474 (ENH-1621). ``completed_date`` is derived from ``completed_at`` (taking the
475 date portion) when present, leaving file-mtime / Resolution-section
476 inference to the file-parsing fallback path.
477 """
478 from little_loops.frontmatter import parse_frontmatter
480 count = 0
481 for issue_file in sorted(issues_dir.rglob("*.md")):
482 try:
483 fm = parse_frontmatter(issue_file.read_text(encoding="utf-8"))
484 except OSError:
485 continue
486 issue_id = fm.get("id")
487 if not issue_id:
488 continue
489 status = str(fm.get("status", "open"))
490 discovered_by = fm.get("discovered_by")
491 captured_at = fm.get("captured_at")
492 completed_at = fm.get("completed_at")
493 ts = str(completed_at or captured_at or fm.get("discovered_date") or "")
494 issue_type, priority = _derive_type_priority(issue_file.name, fm)
495 completed_date: str | None = None
496 if isinstance(completed_at, str) and completed_at:
497 completed_date = completed_at[:10]
498 conn.execute(
499 "INSERT OR IGNORE INTO issue_events("
500 "ts, issue_id, transition, discovered_by, "
501 "issue_type, priority, completed_date, captured_at, completed_at"
502 ") VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
503 (
504 ts,
505 str(issue_id),
506 status,
507 str(discovered_by) if discovered_by else None,
508 issue_type,
509 priority,
510 completed_date,
511 str(captured_at) if captured_at else None,
512 str(completed_at) if completed_at else None,
513 ),
514 )
515 _index(
516 conn,
517 content=f"{issue_id} {status} {issue_type or ''}",
518 kind="issue",
519 ref=str(issue_id),
520 anchor=str(issue_file),
521 ts=ts,
522 )
523 count += 1
524 return count
527def _backfill_loops(conn: sqlite3.Connection, loops_dir: Path) -> int:
528 """Seed ``loop_events`` from FSM state JSON under ``.loops/.running`` + ``.history``."""
529 count = 0
530 for sub in (".running", ".history"):
531 directory = loops_dir / sub
532 if not directory.is_dir():
533 continue
534 for state_file in sorted(directory.glob("*.json")):
535 try:
536 data = json.loads(state_file.read_text(encoding="utf-8"))
537 except (OSError, json.JSONDecodeError):
538 continue
539 if not isinstance(data, dict):
540 continue
541 loop_name = str(data.get("loop_name") or state_file.stem)
542 state = data.get("current_state") or data.get("state")
543 ts = str(data.get("updated_at") or data.get("started_at") or "")
544 conn.execute(
545 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
546 "VALUES(?, ?, ?, ?, ?)",
547 (ts, loop_name, str(state) if state else None, "backfill", None),
548 )
549 _index(
550 conn,
551 content=f"{loop_name} {state or ''}",
552 kind="loop",
553 ref=loop_name,
554 anchor=str(state_file),
555 ts=ts,
556 )
557 count += 1
558 return count
561def _backfill_tool_events(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
562 """Seed ``tool_events`` from assistant tool-use blocks in session JSONL files."""
563 count = 0
564 for jsonl_file in jsonl_files:
565 try:
566 handle = jsonl_file.open(encoding="utf-8")
567 except OSError:
568 continue
569 with handle:
570 for line in handle:
571 line = line.strip()
572 if not line:
573 continue
574 try:
575 record = json.loads(line)
576 except json.JSONDecodeError:
577 continue
578 if record.get("type") != "assistant":
579 continue
580 session_id = record.get("sessionId")
581 ts = str(record.get("timestamp") or "")
582 content = record.get("message", {}).get("content", [])
583 if not isinstance(content, list):
584 continue
585 for block in content:
586 if not isinstance(block, dict) or block.get("type") != "tool_use":
587 continue
588 tool_name = str(block.get("name", ""))
589 args = block.get("input", {})
590 conn.execute(
591 "INSERT INTO tool_events(ts, session_id, tool_name, args_hash, "
592 "result_size, bytes_in, bytes_out, cache_hit) "
593 "VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
594 (ts, session_id, tool_name, _hash_args(args), None, None, None, None),
595 )
596 _index(
597 conn,
598 content=tool_name,
599 kind="tool",
600 ref=tool_name,
601 anchor=str(jsonl_file),
602 ts=ts,
603 )
604 count += 1
605 return count
608def _backfill_messages(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
609 """Seed ``message_events`` from user blocks in session JSONL files.
611 Mirrors :func:`_backfill_tool_events` but selects ``type == "user"`` records
612 and inserts the user's textual content. Used by analyze_workflows() so
613 workflow analysis can read message bodies from the DB instead of a JSONL
614 file (ENH-1621).
615 """
616 count = 0
617 for jsonl_file in jsonl_files:
618 try:
619 handle = jsonl_file.open(encoding="utf-8")
620 except OSError:
621 continue
622 with handle:
623 for line in handle:
624 line = line.strip()
625 if not line:
626 continue
627 try:
628 record = json.loads(line)
629 except json.JSONDecodeError:
630 continue
631 if record.get("type") != "user":
632 continue
633 session_id = record.get("sessionId")
634 ts = str(record.get("timestamp") or "")
635 # The user message body lives at message.content; it may be a
636 # plain string or a list of content blocks. We persist a text
637 # rendering — list blocks are concatenated by their "text"
638 # field so analyze_workflows() can run its regexes over it.
639 content = record.get("message", {}).get("content", "")
640 if isinstance(content, list):
641 parts: list[str] = []
642 for block in content:
643 if isinstance(block, dict) and isinstance(block.get("text"), str):
644 parts.append(block["text"])
645 text = "\n".join(parts)
646 elif isinstance(content, str):
647 text = content
648 else:
649 text = ""
650 if not text.strip():
651 continue
652 conn.execute(
653 "INSERT INTO message_events(ts, session_id, content) VALUES(?, ?, ?)",
654 (ts, str(session_id) if session_id else None, text),
655 )
656 _index(
657 conn,
658 content=text[:512],
659 kind="message",
660 ref=str(session_id) if session_id else "",
661 anchor=str(jsonl_file),
662 ts=ts,
663 )
664 count += 1
665 return count
668def backfill(
669 db: Path | str = DEFAULT_DB_PATH,
670 *,
671 issues_dir: Path | None = None,
672 loops_dir: Path | None = None,
673 jsonl_files: list[Path] | None = None,
674) -> dict[str, int]:
675 """Populate the database from existing on-disk sources.
677 Reads issue-file frontmatter, FSM loop-state JSON, and (optionally) session
678 JSONL tool-use blocks plus user-message blocks. Returns a per-kind count of
679 rows inserted. Sources that are absent are skipped silently.
680 """
681 issues_dir = issues_dir if issues_dir is not None else Path(".issues")
682 loops_dir = loops_dir if loops_dir is not None else Path(".loops")
683 conn = connect(db)
684 counts: dict[str, int] = {"issues": 0, "loops": 0, "tools": 0, "messages": 0}
685 try:
686 if issues_dir.is_dir():
687 counts["issues"] = _backfill_issues(conn, issues_dir)
688 if loops_dir.is_dir():
689 counts["loops"] = _backfill_loops(conn, loops_dir)
690 if jsonl_files:
691 counts["tools"] = _backfill_tool_events(conn, jsonl_files)
692 counts["messages"] = _backfill_messages(conn, jsonl_files)
693 conn.commit()
694 finally:
695 conn.close()
696 return counts