Coverage for little_loops / session_store.py: 15%
936 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:55 -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 backfill_incremental(db,...): incremental JSONL-only backfill filtered by mtime
26 mine_corrections_from_messages(conn,...): scan message_events and insert corrections
27 compact_session(session_id,...): summarize one session into summary_nodes/summary_spans
28 prune(db,...): prune raw event rows older than N days and VACUUM
29 search(db,...): FTS5 full-text query with BM25 ranking
30 recent(db,...): recent rows for a given event kind
31 is_correction(text): return True if text matches a user-correction signal
32 record_correction(db,...): write one row to ``user_corrections`` + search_index
33 record_skill_event(db,...): write one row to ``skill_events`` + search_index
34 cli_event_context(db,...): context manager: INSERT on enter, UPDATE exit_code+duration on exit
35"""
37from __future__ import annotations
39import hashlib
40import json
41import logging
42import os
43import re
44import sqlite3
45import subprocess
46import threading
47import time
48from collections.abc import Generator, Sequence
49from contextlib import contextmanager
50from datetime import UTC, datetime, timedelta
51from pathlib import Path
52from typing import Any
54from little_loops.host_runner import resolve_host
56__all__ = [
57 "DEFAULT_DB_PATH",
58 "SCHEMA_VERSION",
59 "ensure_db",
60 "connect",
61 "SQLiteTransport",
62 "backfill",
63 "backfill_snapshots",
64 "backfill_incremental",
65 "mine_corrections_from_messages",
66 "compact_session",
67 "export_history",
68 "prune",
69 "search",
70 "recent",
71 "is_correction",
72 "record_correction",
73 "record_skill_event",
74 "record_issue_snapshot",
75 "cli_event_context",
76 "resolve_history_db",
77 "record_retirement",
78 "list_retirements",
79]
81logger = logging.getLogger(__name__)
83DEFAULT_DB_PATH = Path(".ll/history.db")
86def resolve_history_db(path: Path | str | None = None) -> Path:
87 """Return the DB path; LL_HISTORY_DB env var takes unconditional precedence."""
88 env_val = os.environ.get("LL_HISTORY_DB")
89 if env_val:
90 return Path(env_val)
91 return Path(path) if path is not None else DEFAULT_DB_PATH
94SCHEMA_VERSION = 14
96_VALID_KINDS = frozenset(
97 {"tool", "file", "issue", "loop", "correction", "message", "skill", "cli", "snapshot"}
98)
99_KIND_TABLE = {
100 "tool": "tool_events",
101 "file": "file_events",
102 "issue": "issue_events",
103 "loop": "loop_events",
104 "correction": "user_corrections",
105 "message": "message_events",
106 "skill": "skill_events",
107 "cli": "cli_events",
108}
110# FSM event types the SQLiteTransport records as loop_events rows.
111_LOOP_EVENT_TYPES = frozenset(
112 {
113 "loop_start",
114 "loop_resume",
115 "loop_complete",
116 "state_enter",
117 "route",
118 "retry_exhausted",
119 "cycle_detected",
120 "max_steps_summary",
121 "max_iterations_reached_summary",
122 }
123)
126_CORRECTION_RE = re.compile(
127 r"^\s*(no[,!]|don'?t\s|stop\s|revert|that'?s\s+wrong|not\s+like\s+that)",
128 re.IGNORECASE,
129)
131_PHRASE_RE = re.compile(
132 r"\b(?:"
133 r"instead"
134 r"|actually\s+(?:that|this|it)\s"
135 r"|you missed"
136 r"|should be\s+(?!fine\b|ok\b|okay\b|good\b|great\b|alright\b|correct\b|right\b)"
137 r"|wrong approach"
138 r"|remember that"
139 r"|always use"
140 r"|never use"
141 r"|from now on"
142 r"|I meant\b.*\bnot\b"
143 r"|not\b.*\buse\b"
144 r")",
145 re.IGNORECASE,
146)
147_REMEMBER_RE = re.compile(r"^!remember\b", re.IGNORECASE)
150def is_correction(text: str, extra_patterns: Sequence[str] = ()) -> bool:
151 """Return True if *text* matches a known user-correction signal.
153 ``extra_patterns`` are raw regex strings appended to the built-in set
154 (``analytics.capture.correction_patterns``). Built-ins always remain active.
155 Invalid patterns are skipped with a warning.
156 """
157 t = text[:512]
158 _extra_re: re.Pattern[str] | None = None
159 if extra_patterns:
160 parts: list[str] = []
161 for p in extra_patterns:
162 try:
163 re.compile(p, re.IGNORECASE)
164 parts.append(f"(?:{p})")
165 except re.error:
166 logger.warning("is_correction: skipping invalid extra_pattern %r", p)
167 if parts:
168 _extra_re = re.compile("|".join(parts), re.IGNORECASE)
169 return bool(
170 _REMEMBER_RE.match(t)
171 or _CORRECTION_RE.match(t)
172 or _PHRASE_RE.search(t)
173 or (_extra_re and _extra_re.search(t))
174 )
177# ---------------------------------------------------------------------------
178# Schema + migrations
179# ---------------------------------------------------------------------------
181# Each entry is the full SQL applied to move the schema from version index to
182# index+1. Migration 0 bootstraps the whole schema; append new entries to
183# evolve it. ``bytes_in`` / ``bytes_out`` / ``cache_hit`` are reserved on
184# ``tool_events`` for FEAT-1160 (Context Window Analytics) so that feature does
185# not require a follow-up migration.
186_MIGRATIONS: list[str] = [
187 """
188 CREATE TABLE IF NOT EXISTS tool_events (
189 id INTEGER PRIMARY KEY AUTOINCREMENT,
190 ts TEXT NOT NULL,
191 session_id TEXT,
192 tool_name TEXT,
193 args_hash TEXT,
194 result_size INTEGER,
195 bytes_in INTEGER,
196 bytes_out INTEGER,
197 cache_hit INTEGER
198 );
199 CREATE TABLE IF NOT EXISTS file_events (
200 id INTEGER PRIMARY KEY AUTOINCREMENT,
201 ts TEXT NOT NULL,
202 session_id TEXT,
203 path TEXT,
204 op TEXT,
205 issue_id TEXT,
206 git_sha TEXT
207 );
208 CREATE TABLE IF NOT EXISTS issue_events (
209 id INTEGER PRIMARY KEY AUTOINCREMENT,
210 ts TEXT NOT NULL,
211 issue_id TEXT,
212 transition TEXT,
213 discovered_by TEXT
214 );
215 CREATE TABLE IF NOT EXISTS loop_events (
216 id INTEGER PRIMARY KEY AUTOINCREMENT,
217 ts TEXT NOT NULL,
218 loop_name TEXT,
219 state TEXT,
220 transition TEXT,
221 retries INTEGER
222 );
223 CREATE TABLE IF NOT EXISTS user_corrections (
224 id INTEGER PRIMARY KEY AUTOINCREMENT,
225 ts TEXT NOT NULL,
226 session_id TEXT,
227 content TEXT,
228 source TEXT
229 );
230 CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
231 content,
232 kind UNINDEXED,
233 ref UNINDEXED,
234 anchor UNINDEXED,
235 ts UNINDEXED
236 );
237 CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);
238 """,
239 # v2 (ENH-1621): widen issue_events with completion-summary columns so
240 # ll-history `summary` can be answered from the DB; add message_events for
241 # analyze_workflows() to read user message bodies without re-parsing JSONL.
242 """
243 ALTER TABLE issue_events ADD COLUMN issue_type TEXT;
244 ALTER TABLE issue_events ADD COLUMN priority TEXT;
245 ALTER TABLE issue_events ADD COLUMN completed_date TEXT;
246 ALTER TABLE issue_events ADD COLUMN captured_at TEXT;
247 ALTER TABLE issue_events ADD COLUMN completed_at TEXT;
248 CREATE TABLE message_events (
249 id INTEGER PRIMARY KEY AUTOINCREMENT,
250 ts TEXT NOT NULL,
251 session_id TEXT,
252 content TEXT
253 );
254 """,
255 # v3 (ENH-1690): unique dedup index on issue_events so INSERT OR IGNORE
256 # prevents duplicate rows when backfill() is called multiple times.
257 """
258 CREATE UNIQUE INDEX IF NOT EXISTS idx_issue_events_dedup
259 ON issue_events(issue_id, transition);
260 """,
261 # v4 (ENH-1710): sessions table maps session_id to its JSONL file path,
262 # closing the broken link between event rows and their source log.
263 """
264 CREATE TABLE sessions (
265 session_id TEXT PRIMARY KEY,
266 jsonl_path TEXT NOT NULL,
267 started_at TEXT,
268 project_path TEXT
269 );
270 """,
271 # v5 (ENH-1711): issue_sessions VIEW joins issue_events to message_events via
272 # overlapping timestamps, making the implicit session→issue link explicit and
273 # queryable. Requires captured_at IS NOT NULL; populated by _backfill_issues()
274 # for historical rows and by issue_lifecycle.py emit sites (ENH-1839) for
275 # live-emitted rows.
276 """
277 CREATE VIEW issue_sessions AS
278 SELECT ie.issue_id,
279 me.session_id,
280 s.jsonl_path,
281 MIN(me.ts) AS first_message_ts,
282 MAX(me.ts) AS last_message_ts
283 FROM issue_events ie
284 JOIN message_events me
285 ON me.ts >= ie.captured_at
286 AND (ie.completed_at IS NULL OR me.ts <= ie.completed_at)
287 LEFT JOIN sessions s ON s.session_id = me.session_id
288 WHERE ie.captured_at IS NOT NULL
289 GROUP BY ie.issue_id, me.session_id;
290 """,
291 # v6 (ENH-1830): last_backfill_ts meta key for incremental JSONL backfill at
292 # session start. The meta table already holds arbitrary key/value pairs; this
293 # initialises the sentinel so reads can distinguish "no prior run" (NULL) from
294 # a real ISO 8601 timestamp string.
295 """
296 INSERT OR IGNORE INTO meta(key, value) VALUES('last_backfill_ts', NULL);
297 """,
298 # v7 (ENH-1833): skill_events table records /ll: skill invocations at dispatch
299 # time via the user_prompt_submit hook so ll-session recent --kind skill works.
300 """
301 CREATE TABLE IF NOT EXISTS skill_events (
302 id INTEGER PRIMARY KEY AUTOINCREMENT,
303 ts TEXT NOT NULL,
304 session_id TEXT,
305 skill_name TEXT,
306 args TEXT
307 );
308 """,
309 # v8 (ENH-1848): cli_events table records ll- CLI invocations via cli_event_context()
310 """
311 CREATE TABLE IF NOT EXISTS cli_events (
312 id INTEGER PRIMARY KEY AUTOINCREMENT,
313 ts TEXT NOT NULL,
314 binary TEXT NOT NULL,
315 args TEXT NOT NULL,
316 exit_code INTEGER,
317 duration_ms INTEGER
318 );
319 """,
320 # v9 (ENH-1904): unique dedup index on user_corrections so INSERT OR IGNORE
321 # enforces idempotency during correction mining. Mirrors v3's
322 # idx_issue_events_dedup pattern.
323 """
324 CREATE UNIQUE INDEX IF NOT EXISTS idx_corrections_dedup
325 ON user_corrections(session_id, content);
326 """,
327 # v10 (FEAT-1712, v12 ENH-1953): LCM-style hierarchical summary DAG over
328 # session history. summary_nodes holds LLM-generated summaries (via three-level
329 # LCM Algorithm 3 escalation: normal → aggressive bullet-point → deterministic
330 # truncation) at multiple levels: 'leaf' nodes cover a fixed token-budget block
331 # of message_events; 'condensed' nodes summarise a session's leaves (level 0,
332 # per-session) or cross-session nodes (level 1+, session_id IS NULL); the root
333 # node sits at the maximum level. summary_spans links summary nodes back to the
334 # originating message_events rows for lossless drill-down.
335 # FK references are decorative (no PRAGMA foreign_keys; integrity enforced at
336 # the application layer by compact_session's insert ordering + INSERT OR IGNORE).
337 # Partial unique indexes prevent duplicate leaf and condensed nodes per session
338 # (idx_summary_nodes_condensed_dedup) and duplicate cross-session nodes
339 # (idx_summary_nodes_cross_dedup, added in v12) across repeated backfill() calls
340 # (idempotency via INSERT OR IGNORE).
341 """
342 CREATE TABLE IF NOT EXISTS summary_nodes (
343 id INTEGER PRIMARY KEY AUTOINCREMENT,
344 kind TEXT NOT NULL,
345 content TEXT NOT NULL,
346 tokens INTEGER,
347 parent_id INTEGER REFERENCES summary_nodes(id),
348 session_id TEXT,
349 ts_start TEXT,
350 ts_end TEXT,
351 created_at TEXT NOT NULL
352 );
353 CREATE TABLE IF NOT EXISTS summary_spans (
354 summary_id INTEGER REFERENCES summary_nodes(id),
355 message_event_id INTEGER REFERENCES message_events(id),
356 PRIMARY KEY (summary_id, message_event_id)
357 );
358 CREATE UNIQUE INDEX IF NOT EXISTS idx_summary_nodes_leaf_dedup
359 ON summary_nodes(session_id, ts_start, ts_end) WHERE kind = 'leaf';
360 CREATE UNIQUE INDEX IF NOT EXISTS idx_summary_nodes_condensed_dedup
361 ON summary_nodes(session_id) WHERE kind = 'condensed';
362 CREATE INDEX IF NOT EXISTS idx_summary_nodes_parent_id
363 ON summary_nodes(parent_id);
364 """,
365 # v11 (ENH-1942): assistant_messages stores concatenated text blocks from
366 # assistant responses so the SFT pipeline can read conversation turn-pairs
367 # from the database instead of re-parsing JSONL. tool_use_count enables
368 # filter predicates (e.g. min_tool_invocations) without a JOIN.
369 # idx_assistant_messages_dedup mirrors v3's idx_issue_events_dedup pattern
370 # so INSERT OR IGNORE enforces idempotency during backfill.
371 """
372 CREATE TABLE IF NOT EXISTS assistant_messages (
373 id INTEGER PRIMARY KEY AUTOINCREMENT,
374 ts TEXT NOT NULL,
375 session_id TEXT NOT NULL,
376 content TEXT NOT NULL,
377 tool_use_count INTEGER DEFAULT 0
378 );
379 CREATE INDEX IF NOT EXISTS idx_assistant_messages_session_ts
380 ON assistant_messages(session_id, ts);
381 CREATE UNIQUE INDEX IF NOT EXISTS idx_assistant_messages_dedup
382 ON assistant_messages(session_id, ts, content);
383 """,
384 # v12 (ENH-1953): add level column to summary_nodes for N-level DAG
385 # traversal and a cross-session dedup index. level 0 = leaf/per-session
386 # condensed, 1+ = cross-session condensed, max = root.
387 # idx_summary_nodes_cross_dedup prevents duplicate cross-session condensed
388 # nodes where session_id IS NULL (the existing idx_summary_nodes_condensed_dedup
389 # only covers per-session rows and is unchanged).
390 """
391 ALTER TABLE summary_nodes ADD COLUMN level INTEGER DEFAULT 0;
392 CREATE UNIQUE INDEX IF NOT EXISTS idx_summary_nodes_cross_dedup
393 ON summary_nodes(level, ts_start, ts_end)
394 WHERE kind='condensed' AND session_id IS NULL;
395 """,
396 # v13 (ENH-2046): correction_retirements — records addressed correction clusters so
397 # detect_recurring_feedback() excludes already-ruled topics from future runs.
398 """
399 CREATE TABLE IF NOT EXISTS correction_retirements (
400 id INTEGER PRIMARY KEY AUTOINCREMENT,
401 topic_fingerprint TEXT NOT NULL,
402 rule_id TEXT,
403 addressed_at TEXT NOT NULL,
404 session_id TEXT
405 );
406 CREATE UNIQUE INDEX IF NOT EXISTS idx_retirements_fingerprint
407 ON correction_retirements(topic_fingerprint);
408 """,
409 # v14 (ENH-2151): issue_snapshots — stores full issue content at key lifecycle
410 # transitions (captured, done, cancelled) so completed issue context is queryable
411 # from the DB even after the source .md file is moved or deleted.
412 # FTS via the existing autonomous search_index with kind="snapshot" (Decision 1).
413 # Dedup index mirrors v3's idx_issue_events_dedup pattern.
414 """
415 CREATE TABLE IF NOT EXISTS issue_snapshots (
416 id INTEGER PRIMARY KEY AUTOINCREMENT,
417 ts TEXT NOT NULL,
418 issue_id TEXT NOT NULL,
419 transition TEXT NOT NULL,
420 title TEXT,
421 priority TEXT,
422 issue_type TEXT,
423 body TEXT,
424 frontmatter TEXT
425 );
426 CREATE UNIQUE INDEX IF NOT EXISTS idx_issue_snapshots_dedup
427 ON issue_snapshots(issue_id, transition);
428 """,
429]
432def _now() -> str:
433 """Return the current UTC time as a Z-suffixed ISO 8601 string."""
434 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
437# Milliseconds a contended open will wait for the write lock before giving up.
438# Every ``ll-*`` invocation opens this DB on startup; under ll-auto / ll-loop /
439# ll-parallel many processes contend at once, so without a busy_timeout an open
440# fails instantly with ``OperationalError: database is locked``.
441_BUSY_TIMEOUT_MS = 5000
444def _configure_connection(conn: sqlite3.Connection) -> None:
445 """Apply concurrency pragmas to a freshly opened connection.
447 ``busy_timeout`` makes a contended open wait instead of failing instantly
448 with ``database is locked``; WAL journal mode lets readers and writers
449 proceed concurrently — critical for the multi-process ll-auto / ll-loop /
450 ll-parallel workload, where rollback-journal mode otherwise serialises every
451 reader behind any active writer. WAL is a persistent property of the database
452 file, so re-applying it on every connection is idempotent. Both pragmas are
453 best-effort: a read-only filesystem or an older SQLite that rejects one must
454 not prevent the database from opening.
455 """
456 try:
457 conn.execute(f"PRAGMA busy_timeout = {_BUSY_TIMEOUT_MS}")
458 conn.execute("PRAGMA journal_mode = WAL")
459 except sqlite3.OperationalError:
460 logger.debug("session_store: could not apply connection pragmas", exc_info=True)
463def _split_sql_statements(script: str) -> list[str]:
464 """Split a migration's SQL into individual statements on ``;`` boundaries.
466 Used instead of :meth:`sqlite3.Connection.executescript` because the latter
467 issues an implicit ``COMMIT`` that would release the write lock held across
468 the migration sequence (see :func:`_apply_migrations`). The migration SQL in
469 ``_MIGRATIONS`` is fully controlled and contains no semicolons inside string
470 literals or column definitions, so a plain ``;`` split is safe here; do not
471 repurpose this for arbitrary user SQL.
472 """
473 return [stmt for raw in script.split(";") if (stmt := raw.strip())]
476def _current_version(conn: sqlite3.Connection) -> int:
477 """Return the applied schema version, or 0 if the meta table is absent.
479 Only a genuinely missing ``meta`` table means "fresh database, version 0".
480 A transient ``database is locked`` (another process mid-write) is a different
481 ``OperationalError`` and must propagate — misreading it as 0 makes the caller
482 re-run migration 0 and crash with "table tool_events already exists".
483 """
484 try:
485 row = conn.execute("SELECT value FROM meta WHERE key = 'schema_version'").fetchone()
486 except sqlite3.OperationalError as exc:
487 if "no such table" in str(exc).lower():
488 return 0
489 raise
490 return int(row[0]) if row else 0
493def _apply_migrations(conn: sqlite3.Connection) -> None:
494 """Apply every migration newer than the database's current version.
496 The whole sequence runs inside a single ``BEGIN IMMEDIATE`` transaction so
497 concurrent processes serialise: the first to acquire the write lock migrates
498 while the rest wait (``busy_timeout``), then re-read the now-current version
499 and apply nothing. The version is re-checked *inside* the lock to close the
500 fresh-database race where two processes both read version 0 and both try to
501 create the bootstrap tables. ``executescript`` is avoided because its implicit
502 leading ``COMMIT`` would drop the lock between migrations.
504 Fast path: when the schema is already current, return without taking the
505 write lock at all — in WAL mode this read never blocks on a concurrent
506 writer, so the steady-state ``ll-*`` startup stays lock-free.
507 """
508 if _current_version(conn) >= len(_MIGRATIONS):
509 return
510 prior_isolation = conn.isolation_level
511 conn.isolation_level = None # manual transaction control
512 try:
513 conn.execute("BEGIN IMMEDIATE")
514 try:
515 version = _current_version(conn)
516 for index in range(version, len(_MIGRATIONS)):
517 for statement in _split_sql_statements(_MIGRATIONS[index]):
518 conn.execute(statement)
519 conn.execute(
520 "INSERT INTO meta(key, value) VALUES('schema_version', ?) "
521 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
522 (str(index + 1),),
523 )
524 conn.execute("COMMIT")
525 except BaseException:
526 conn.execute("ROLLBACK")
527 raise
528 finally:
529 conn.isolation_level = prior_isolation
532def ensure_db(path: Path | str = DEFAULT_DB_PATH) -> Path:
533 """Create the database at *path* (if needed) and apply pending migrations.
535 Idempotent: safe to call on every session start. The parent directory is
536 created if absent. Returns the resolved database path.
538 On the first call after the ENH-1635 rename, transparently migrates a
539 pre-existing ``.ll/session.db`` (and any ``-shm``/``-wal`` sidecars) to
540 the new ``.ll/history.db`` path. Each sidecar is renamed independently so
541 a single failure does not abort the others; failures are logged at
542 WARNING (the caller in ``hooks/session_start.py`` wraps the whole call
543 in ``contextlib.suppress(Exception)``, which would otherwise silence
544 diagnostic context).
545 """
546 db_path = Path(path)
547 if db_path == DEFAULT_DB_PATH:
548 env_val = os.environ.get("LL_HISTORY_DB")
549 if env_val:
550 db_path = Path(env_val)
551 legacy = db_path.parent / "session.db"
552 if legacy.exists() and not db_path.exists():
553 for suffix in ("", "-shm", "-wal"):
554 src = legacy.parent / f"session.db{suffix}"
555 if src.exists():
556 dst = db_path.parent / f"history.db{suffix}"
557 try:
558 src.rename(dst)
559 logger.info("session_store: migrated %s -> %s", src, dst)
560 except OSError:
561 logger.warning(
562 "session_store: legacy rename failed for %s; continuing with fresh db",
563 src,
564 exc_info=True,
565 )
566 break
567 db_path.parent.mkdir(parents=True, exist_ok=True)
568 conn = sqlite3.connect(str(db_path))
569 try:
570 _configure_connection(conn)
571 _apply_migrations(conn)
572 finally:
573 conn.close()
574 return db_path
577def connect(path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
578 """Open a connection to the session database, ensuring the schema first.
580 Rows are returned as :class:`sqlite3.Row` so callers can index by name.
581 """
582 db_path = ensure_db(path)
583 conn = sqlite3.connect(str(db_path))
584 _configure_connection(conn)
585 conn.row_factory = sqlite3.Row
586 return conn
589def _index(
590 conn: sqlite3.Connection,
591 *,
592 content: str,
593 kind: str,
594 ref: str,
595 anchor: str,
596 ts: str,
597) -> None:
598 """Insert one row into the FTS5 ``search_index`` table."""
599 conn.execute(
600 "INSERT INTO search_index(content, kind, ref, anchor, ts) VALUES(?, ?, ?, ?, ?)",
601 (content, kind, ref, anchor, ts),
602 )
605def write_file_event(
606 db_path: Path | str,
607 session_id: str | None,
608 path: str,
609 op: str,
610 issue_id: str | None = None,
611 config: dict | None = None,
612) -> None:
613 """Write one row to ``file_events`` and index it in ``search_index``.
615 Gated by ``analytics.capture.file_events`` (ENH-1841): when ``config`` is
616 provided and the flag is ``false``, the write is suppressed. Missing ``capture``
617 key defaults to permissive (no behavior change).
618 """
619 if config is not None:
620 from little_loops.config.features import AnalyticsCaptureConfig
622 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
623 if not capture.file_events:
624 return
625 conn = connect(db_path)
626 ts = _now()
627 try:
628 conn.execute(
629 "INSERT INTO file_events(ts, session_id, path, op, issue_id, git_sha) "
630 "VALUES(?, ?, ?, ?, ?, ?)",
631 (ts, session_id, path, op, issue_id, None),
632 )
633 _index(conn, content=path, kind="file", ref=path, anchor=op, ts=ts)
634 conn.commit()
635 finally:
636 conn.close()
639def record_correction(
640 db_path: Path | str,
641 session_id: str | None,
642 content: str,
643 source: str,
644 config: dict | None = None,
645) -> None:
646 """Write one row to ``user_corrections`` and index it in ``search_index``.
648 Gated by ``analytics.capture.corrections`` (ENH-1841): when ``config`` is
649 provided and the flag is ``false``, the write is suppressed. Missing ``capture``
650 key defaults to permissive (no behavior change).
651 """
652 if config is not None:
653 from little_loops.config.features import AnalyticsCaptureConfig
655 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
656 if not capture.corrections:
657 return
658 content = content[:512]
659 conn = connect(db_path)
660 ts = _now()
661 try:
662 conn.execute(
663 "INSERT INTO user_corrections(ts, session_id, content, source) VALUES(?, ?, ?, ?)",
664 (ts, session_id, content, source),
665 )
666 _index(conn, content=content, kind="correction", ref=session_id or "", anchor=source, ts=ts)
667 conn.commit()
668 finally:
669 conn.close()
672def record_skill_event(
673 db_path: Path | str,
674 session_id: str | None,
675 skill_name: str,
676 args: str,
677 config: dict | None = None,
678) -> None:
679 """Write one row to ``skill_events`` and index it in ``search_index``.
681 The ``config`` parameter is a forward-compatibility stub for ENH-1835, which
682 will inject a per-skill analytics gate without changing this signature.
683 """
684 args = args[:200]
685 conn = connect(db_path)
686 ts = _now()
687 try:
688 conn.execute(
689 "INSERT INTO skill_events(ts, session_id, skill_name, args) VALUES(?, ?, ?, ?)",
690 (ts, session_id, skill_name, args),
691 )
692 _index(
693 conn, content=skill_name, kind="skill", ref=session_id or "", anchor=skill_name, ts=ts
694 )
695 conn.commit()
696 finally:
697 conn.close()
700def record_issue_snapshot(
701 db_path: Path | str,
702 issue_id: str,
703 transition: str,
704 file_path: str,
705) -> None:
706 """Write one row to ``issue_snapshots`` and index it in ``search_index``.
708 Reads the issue file at *file_path*, extracts frontmatter metadata and
709 markdown body, then inserts into ``issue_snapshots`` using ``INSERT OR IGNORE``
710 so repeated calls for the same ``(issue_id, transition)`` are idempotent.
711 Also calls ``_index()`` with ``kind="snapshot"`` so FTS5 searches surface it.
713 Silently returns if *file_path* does not exist or cannot be read.
714 """
715 from little_loops.frontmatter import parse_frontmatter, strip_frontmatter
717 try:
718 content = Path(file_path).read_text(encoding="utf-8")
719 except OSError:
720 return
722 fm = parse_frontmatter(content)
723 body = strip_frontmatter(content)
724 title = fm.get("title") or fm.get("id") or issue_id
725 priority = fm.get("priority")
726 issue_type = fm.get("type")
728 # Serialise frontmatter as JSON for storage.
729 fm_json = json.dumps({k: str(v) for k, v in fm.items() if v is not None}, sort_keys=True)
731 conn = connect(db_path)
732 ts = _now()
733 try:
734 conn.execute(
735 "INSERT OR IGNORE INTO issue_snapshots"
736 "(ts, issue_id, transition, title, priority, issue_type, body, frontmatter)"
737 " VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
738 (ts, issue_id, transition, str(title), priority, issue_type, body, fm_json),
739 )
740 _index(
741 conn,
742 content=f"{issue_id} {title} {body or ''}".strip(),
743 kind="snapshot",
744 ref=issue_id,
745 anchor=file_path,
746 ts=ts,
747 )
748 conn.commit()
749 finally:
750 conn.close()
753@contextmanager
754def cli_event_context(
755 db_path: Path | str = DEFAULT_DB_PATH,
756 binary: str = "",
757 args: list[str] | None = None,
758 config: dict | None = None,
759) -> Generator[None, None, None]:
760 """Insert a ``cli_events`` row on enter; update exit_code and duration_ms on exit.
762 The ``config`` parameter is a forward-compatibility stub for ENH-1835 gating;
763 it is accepted but not yet used.
764 """
765 if args is None:
766 args = []
767 effective_path = (
768 resolve_history_db(db_path) if Path(db_path) == DEFAULT_DB_PATH else Path(db_path)
769 )
770 conn = connect(effective_path)
771 start = time.time()
772 ts = _now()
773 cursor = conn.execute(
774 "INSERT INTO cli_events(ts, binary, args) VALUES(?, ?, ?)",
775 (ts, binary, json.dumps(args[:50])),
776 )
777 row_id = cursor.lastrowid
778 conn.commit()
779 exit_code = 0
780 try:
781 yield
782 except BaseException:
783 exit_code = 1
784 raise
785 finally:
786 duration_ms = int((time.time() - start) * 1000)
787 conn.execute(
788 "UPDATE cli_events SET exit_code=?, duration_ms=? WHERE id=?",
789 (exit_code, duration_ms, row_id),
790 )
791 conn.commit()
792 conn.close()
795# ---------------------------------------------------------------------------
796# Query API
797# ---------------------------------------------------------------------------
800def search(
801 db: Path | str = DEFAULT_DB_PATH,
802 *,
803 query: str,
804 limit: int = 20,
805) -> list[dict[str, Any]]:
806 """Run an FTS5 full-text query, returning BM25-ranked results.
808 Each result dict carries ``content``, ``kind``, ``ref``, ``anchor`` (a
809 file:line-style pointer where available), ``ts`` and a numeric ``score``
810 (lower BM25 score = better match).
811 """
812 conn = connect(db)
813 try:
814 rows = conn.execute(
815 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
816 "FROM search_index WHERE search_index MATCH ? "
817 "ORDER BY score LIMIT ?",
818 (query, limit),
819 ).fetchall()
820 except sqlite3.OperationalError as exc:
821 raise ValueError(f"invalid FTS query {query!r}: {exc}") from exc
822 finally:
823 conn.close()
824 return [dict(row) for row in rows]
827def recent(
828 db: Path | str = DEFAULT_DB_PATH,
829 *,
830 kind: str,
831 limit: int = 20,
832) -> list[dict[str, Any]]:
833 """Return the most recent rows for *kind* (tool, file, issue, loop, correction, message, skill, cli)."""
834 if kind not in _VALID_KINDS:
835 raise ValueError(f"unknown kind {kind!r}; expected one of {sorted(_VALID_KINDS)}")
836 table = _KIND_TABLE[kind]
837 conn = connect(db)
838 try:
839 rows = conn.execute(
840 f"SELECT * FROM {table} ORDER BY id DESC LIMIT ?", # noqa: S608 - table from fixed map
841 (limit,),
842 ).fetchall()
843 finally:
844 conn.close()
845 return [dict(row) for row in rows]
848# ---------------------------------------------------------------------------
849# SQLiteTransport
850# ---------------------------------------------------------------------------
852_ISSUE_TRANSITION_MAP: dict[str, str] = {
853 "issue.completed": "done",
854 "issue.closed": "done",
855 "issue.deferred": "deferred",
856 "issue.skipped": "cancelled",
857 "issue.created": "open",
858 "issue.started": "in_progress",
859}
862def _derive_transition(event_type: str) -> str:
863 """Map an ``issue.*`` event type to the canonical transition/status string."""
864 return _ISSUE_TRANSITION_MAP.get(event_type, event_type.split(".", 1)[1])
867class SQLiteTransport:
868 """EventBus sink that records FSM loop events into the session database.
870 A single connection is opened at construction with ``check_same_thread``
871 disabled, since :meth:`send` may be called from the FSM thread while other
872 transports run their own threads; a lock serialises writes. Every
873 operation is best-effort — a database error is logged and swallowed so a
874 failing sink never aborts a loop run (the four ``wire_transports`` call
875 sites depend on this).
876 """
878 def __init__(self, db_path: Path | str = DEFAULT_DB_PATH) -> None:
879 resolved = Path(db_path)
880 if resolved == DEFAULT_DB_PATH:
881 env_val = os.environ.get("LL_HISTORY_DB")
882 if env_val:
883 resolved = Path(env_val)
884 self._path = resolved
885 self._lock = threading.Lock()
886 self._conn: sqlite3.Connection | None = None
887 try:
888 ensure_db(self._path)
889 self._conn = sqlite3.connect(str(self._path), check_same_thread=False)
890 _configure_connection(self._conn)
891 except sqlite3.Error:
892 logger.warning(
893 "SQLiteTransport: could not open %s; sink disabled", self._path, exc_info=True
894 )
895 self._conn = None
897 def send(self, event: dict[str, Any]) -> None:
898 """Record a recognised event as a ``loop_events`` or ``issue_events`` row (best-effort)."""
899 conn = self._conn
900 if conn is None:
901 return
902 event_type = str(event.get("event", ""))
903 ts = str(event.get("ts") or _now())
904 try:
905 with self._lock:
906 if event_type in _LOOP_EVENT_TYPES:
907 loop_name = str(event.get("loop_name", "")) or None
908 state = event.get("state")
909 if event_type == "loop_complete":
910 state = event.get("outcome", state)
911 retries = event.get("retries")
912 conn.execute(
913 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
914 "VALUES(?, ?, ?, ?, ?)",
915 (
916 ts,
917 loop_name,
918 str(state) if state is not None else None,
919 event_type,
920 int(retries) if isinstance(retries, int) else None,
921 ),
922 )
923 _index(
924 conn,
925 content=" ".join(
926 str(p) for p in (loop_name, state, event_type) if p is not None
927 ),
928 kind="loop",
929 ref=loop_name or "",
930 anchor=f".loops/{loop_name}.yaml" if loop_name else "",
931 ts=ts,
932 )
933 elif event_type.startswith("issue."):
934 issue_id = event.get("issue_id")
935 transition = _derive_transition(event_type)
936 conn.execute(
937 "INSERT OR IGNORE INTO issue_events("
938 "ts, issue_id, transition, discovered_by, "
939 "issue_type, priority, captured_at, completed_at"
940 ") VALUES(?,?,?,?,?,?,?,?)",
941 (
942 ts,
943 issue_id,
944 transition,
945 event.get("discovered_by"),
946 event.get("issue_type"),
947 event.get("priority"),
948 event.get("captured_at"),
949 event.get("completed_at"),
950 ),
951 )
952 _index(
953 conn,
954 content=f"{issue_id or ''} {event.get('issue_type', '')}".strip(),
955 kind="issue",
956 ref=str(issue_id or ""),
957 anchor=event.get("issue_file", ""),
958 ts=ts,
959 )
960 # Side-effect: write content snapshot when the event carries a file path.
961 file_path = event.get("file_path")
962 if file_path and issue_id and transition in ("done", "open", "cancelled"):
963 try:
964 conn.commit() # flush issue_events before spawning new conn
965 except sqlite3.Error:
966 pass
967 record_issue_snapshot(self._path, str(issue_id), transition, str(file_path))
968 return # skip second commit below; record_issue_snapshot committed
969 else:
970 return
971 conn.commit()
972 except sqlite3.Error:
973 logger.warning("SQLiteTransport: write failed for event %r", event_type, exc_info=True)
975 def close(self) -> None:
976 """Close the underlying connection (best-effort)."""
977 if self._conn is not None:
978 try:
979 self._conn.close()
980 except sqlite3.Error:
981 pass
982 self._conn = None
985# ---------------------------------------------------------------------------
986# Backfill
987# ---------------------------------------------------------------------------
990def _hash_args(value: Any) -> str:
991 """Return a short stable hash of a tool-call argument structure."""
992 try:
993 blob = json.dumps(value, sort_keys=True, default=str)
994 except (TypeError, ValueError):
995 blob = repr(value)
996 return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
999_FILENAME_TYPE_RE = re.compile(r"(BUG|ENH|FEAT|EPIC)-(\d+)")
1000_FILENAME_PRIORITY_RE = re.compile(r"^(P\d)")
1003def _derive_type_priority(filename: str, fm: dict[str, Any]) -> tuple[str | None, str | None]:
1004 """Derive (issue_type, priority) preferring frontmatter, falling back to filename.
1006 Mirrors :func:`little_loops.issue_history.parsing.parse_completed_issue`'s
1007 filename-parsing convention (``P[0-5]-[TYPE]-[NNN]-...``).
1008 """
1009 fm_type = fm.get("type")
1010 issue_type: str | None = str(fm_type) if isinstance(fm_type, str) and fm_type else None
1011 fm_priority = fm.get("priority")
1012 priority: str | None = (
1013 str(fm_priority) if isinstance(fm_priority, str) and fm_priority else None
1014 )
1015 if issue_type is None:
1016 m = _FILENAME_TYPE_RE.search(filename)
1017 if m:
1018 issue_type = m.group(1)
1019 if priority is None:
1020 m = _FILENAME_PRIORITY_RE.match(filename)
1021 if m:
1022 priority = m.group(1)
1023 return issue_type, priority
1026def _backfill_issues(conn: sqlite3.Connection, issues_dir: Path) -> int:
1027 """Seed ``issue_events`` from issue-file frontmatter under *issues_dir*.
1029 Populates the v2 summary columns (``issue_type``, ``priority``,
1030 ``completed_date``, ``captured_at``, ``completed_at``) so ``ll-history
1031 summary`` can be answered from the DB without re-reading the files
1032 (ENH-1621). ``completed_date`` is derived from ``completed_at`` (taking the
1033 date portion) when present, leaving file-mtime / Resolution-section
1034 inference to the file-parsing fallback path.
1035 """
1036 from little_loops.frontmatter import parse_frontmatter
1038 count = 0
1039 for issue_file in sorted(issues_dir.rglob("*.md")):
1040 try:
1041 fm = parse_frontmatter(issue_file.read_text(encoding="utf-8"))
1042 except OSError:
1043 continue
1044 issue_id = fm.get("id")
1045 if not issue_id:
1046 m = _FILENAME_TYPE_RE.search(issue_file.name)
1047 if m:
1048 issue_id = f"{m.group(1)}-{m.group(2)}"
1049 if not issue_id:
1050 continue
1051 status = str(fm.get("status", "open"))
1052 discovered_by = fm.get("discovered_by")
1053 captured_at = fm.get("captured_at")
1054 completed_at = fm.get("completed_at")
1055 ts = str(completed_at or captured_at or fm.get("discovered_date") or "")
1056 issue_type, priority = _derive_type_priority(issue_file.name, fm)
1057 completed_date: str | None = None
1058 if isinstance(completed_at, str) and completed_at:
1059 completed_date = completed_at[:10]
1060 conn.execute(
1061 "INSERT OR IGNORE INTO issue_events("
1062 "ts, issue_id, transition, discovered_by, "
1063 "issue_type, priority, completed_date, captured_at, completed_at"
1064 ") VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
1065 (
1066 ts,
1067 str(issue_id),
1068 status,
1069 str(discovered_by) if discovered_by else None,
1070 issue_type,
1071 priority,
1072 completed_date,
1073 str(captured_at) if captured_at else None,
1074 str(completed_at) if completed_at else None,
1075 ),
1076 )
1077 _index(
1078 conn,
1079 content=f"{issue_id} {status} {issue_type or ''}",
1080 kind="issue",
1081 ref=str(issue_id),
1082 anchor=str(issue_file),
1083 ts=ts,
1084 )
1085 count += 1
1086 return count
1089def _backfill_snapshots(conn: sqlite3.Connection, issues_dir: Path) -> int:
1090 """Seed ``issue_snapshots`` and ``search_index`` from issue files under *issues_dir*.
1092 Follows the ``_backfill_issues()`` pattern: iterates ``*.md`` files, reads
1093 frontmatter + body, inserts with ``INSERT OR IGNORE`` for idempotency.
1094 Uses the issue's current ``status`` as the ``transition`` value.
1095 """
1096 from little_loops.frontmatter import parse_frontmatter, strip_frontmatter
1098 count = 0
1099 ts = _now()
1100 for issue_file in sorted(issues_dir.rglob("*.md")):
1101 try:
1102 content = issue_file.read_text(encoding="utf-8")
1103 except OSError:
1104 continue
1105 fm = parse_frontmatter(content)
1106 issue_id = fm.get("id")
1107 if not issue_id:
1108 m = _FILENAME_TYPE_RE.search(issue_file.name)
1109 if m:
1110 issue_id = f"{m.group(1)}-{m.group(2)}"
1111 if not issue_id:
1112 continue
1113 transition = str(fm.get("status", "open"))
1114 title = fm.get("title") or fm.get("id") or issue_id
1115 priority = fm.get("priority")
1116 issue_type = fm.get("type")
1117 body = strip_frontmatter(content)
1118 fm_json = json.dumps({k: str(v) for k, v in fm.items() if v is not None}, sort_keys=True)
1119 conn.execute(
1120 "INSERT OR IGNORE INTO issue_snapshots"
1121 "(ts, issue_id, transition, title, priority, issue_type, body, frontmatter)"
1122 " VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
1123 (ts, str(issue_id), transition, str(title), priority, issue_type, body, fm_json),
1124 )
1125 _index(
1126 conn,
1127 content=f"{issue_id} {title} {body or ''}".strip(),
1128 kind="snapshot",
1129 ref=str(issue_id),
1130 anchor=str(issue_file),
1131 ts=ts,
1132 )
1133 count += 1
1134 return count
1137def _backfill_loops(conn: sqlite3.Connection, loops_dir: Path) -> int:
1138 """Seed ``loop_events`` from FSM state JSON under ``.loops/.running`` + ``.history``."""
1139 count = 0
1140 for sub in (".running", ".history"):
1141 directory = loops_dir / sub
1142 if not directory.is_dir():
1143 continue
1144 for state_file in sorted(directory.glob("*.json")):
1145 try:
1146 data = json.loads(state_file.read_text(encoding="utf-8"))
1147 except (OSError, json.JSONDecodeError):
1148 continue
1149 if not isinstance(data, dict):
1150 continue
1151 loop_name = str(data.get("loop_name") or state_file.stem)
1152 state = data.get("current_state") or data.get("state")
1153 ts = str(data.get("updated_at") or data.get("started_at") or "")
1154 conn.execute(
1155 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
1156 "VALUES(?, ?, ?, ?, ?)",
1157 (ts, loop_name, str(state) if state else None, "backfill", None),
1158 )
1159 _index(
1160 conn,
1161 content=f"{loop_name} {state or ''}",
1162 kind="loop",
1163 ref=loop_name,
1164 anchor=str(state_file),
1165 ts=ts,
1166 )
1167 count += 1
1168 return count
1171def _backfill_tool_events(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1172 """Seed ``tool_events`` from assistant tool-use blocks in session JSONL files."""
1173 count = 0
1174 for jsonl_file in jsonl_files:
1175 try:
1176 handle = jsonl_file.open(encoding="utf-8")
1177 except OSError:
1178 continue
1179 with handle:
1180 for line in handle:
1181 line = line.strip()
1182 if not line:
1183 continue
1184 try:
1185 record = json.loads(line)
1186 except json.JSONDecodeError:
1187 continue
1188 if record.get("type") != "assistant":
1189 continue
1190 session_id = record.get("sessionId")
1191 ts = str(record.get("timestamp") or "")
1192 content = record.get("message", {}).get("content", [])
1193 if not isinstance(content, list):
1194 continue
1195 for block in content:
1196 if not isinstance(block, dict) or block.get("type") != "tool_use":
1197 continue
1198 tool_name = str(block.get("name", ""))
1199 args = block.get("input", {})
1200 conn.execute(
1201 "INSERT INTO tool_events(ts, session_id, tool_name, args_hash, "
1202 "result_size, bytes_in, bytes_out, cache_hit) "
1203 "VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
1204 (ts, session_id, tool_name, _hash_args(args), None, None, None, None),
1205 )
1206 _index(
1207 conn,
1208 content=tool_name,
1209 kind="tool",
1210 ref=tool_name,
1211 anchor=str(jsonl_file),
1212 ts=ts,
1213 )
1214 count += 1
1215 return count
1218def _backfill_messages(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1219 """Seed ``message_events`` from user blocks in session JSONL files.
1221 Mirrors :func:`_backfill_tool_events` but selects ``type == "user"`` records
1222 and inserts the user's textual content. Used by analyze_workflows() so
1223 workflow analysis can read message bodies from the DB instead of a JSONL
1224 file (ENH-1621).
1225 """
1226 count = 0
1227 for jsonl_file in jsonl_files:
1228 try:
1229 handle = jsonl_file.open(encoding="utf-8")
1230 except OSError:
1231 continue
1232 with handle:
1233 for line in handle:
1234 line = line.strip()
1235 if not line:
1236 continue
1237 try:
1238 record = json.loads(line)
1239 except json.JSONDecodeError:
1240 continue
1241 if record.get("type") != "user":
1242 continue
1243 session_id = record.get("sessionId")
1244 ts = str(record.get("timestamp") or "")
1245 # The user message body lives at message.content; it may be a
1246 # plain string or a list of content blocks. We persist a text
1247 # rendering — list blocks are concatenated by their "text"
1248 # field so analyze_workflows() can run its regexes over it.
1249 content = record.get("message", {}).get("content", "")
1250 if isinstance(content, list):
1251 parts: list[str] = []
1252 for block in content:
1253 if isinstance(block, dict) and isinstance(block.get("text"), str):
1254 parts.append(block["text"])
1255 text = "\n".join(parts)
1256 elif isinstance(content, str):
1257 text = content
1258 else:
1259 text = ""
1260 if not text.strip():
1261 continue
1262 conn.execute(
1263 "INSERT INTO message_events(ts, session_id, content) VALUES(?, ?, ?)",
1264 (ts, str(session_id) if session_id else None, text),
1265 )
1266 _index(
1267 conn,
1268 content=text[:512],
1269 kind="message",
1270 ref=str(session_id) if session_id else "",
1271 anchor=str(jsonl_file),
1272 ts=ts,
1273 )
1274 count += 1
1275 return count
1278def _backfill_assistant_messages(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1279 """Seed ``assistant_messages`` from assistant blocks in session JSONL files.
1281 Mirrors :func:`_backfill_messages` but selects ``type == "assistant"`` records
1282 and concatenates text blocks with ``"\\n\\n"`` — matching the output shape of
1283 ``_extract_turn_pairs()`` in ``user_messages.py``. Also counts ``tool_use``
1284 blocks and stores the count in ``tool_use_count`` so filter predicates like
1285 ``min_tool_invocations`` (ENH-1941) can operate without a JOIN.
1287 Idempotent: INSERT OR IGNORE prevents duplicate rows on repeated backfill.
1288 Depends on the ``sessions`` table (v4 / ENH-1710) for the session_id→JSONL
1289 mapping used by ``conversation_turns()`` to JOIN on session_id.
1290 """
1291 count = 0
1292 for jsonl_file in jsonl_files:
1293 try:
1294 handle = jsonl_file.open(encoding="utf-8")
1295 except OSError:
1296 continue
1297 session_id_from_file: str | None = None
1298 with handle:
1299 for line in handle:
1300 line = line.strip()
1301 if not line:
1302 continue
1303 try:
1304 record = json.loads(line)
1305 except json.JSONDecodeError:
1306 continue
1307 if record.get("type") != "assistant":
1308 continue
1309 session_id = record.get("sessionId")
1310 if session_id_from_file is None and session_id:
1311 session_id_from_file = str(session_id)
1312 ts = str(record.get("timestamp") or "")
1313 content = record.get("message", {}).get("content", [])
1314 if not isinstance(content, list):
1315 continue
1316 # Collect text blocks and count tool_use blocks
1317 text_blocks: list[str] = []
1318 tool_use_count = 0
1319 for block in content:
1320 if isinstance(block, dict):
1321 if block.get("type") == "text":
1322 txt = block.get("text", "").strip()
1323 if txt:
1324 text_blocks.append(txt)
1325 elif block.get("type") == "tool_use":
1326 tool_use_count += 1
1327 if not text_blocks:
1328 continue
1329 concatenated = "\n\n".join(text_blocks)
1330 conn.execute(
1331 "INSERT OR IGNORE INTO assistant_messages(ts, session_id, content, tool_use_count)"
1332 " VALUES(?, ?, ?, ?)",
1333 (ts, str(session_id) if session_id else None, concatenated, tool_use_count),
1334 )
1335 _index(
1336 conn,
1337 content=concatenated[:512],
1338 kind="message",
1339 ref=str(session_id) if session_id else "",
1340 anchor=str(jsonl_file),
1341 ts=ts,
1342 )
1343 count += 1
1344 return count
1347_BACKFILL_SKILL_RE = re.compile(r"<command-name>/ll:(\S+)")
1348_BACKFILL_ARGS_RE = re.compile(r"<command-args>(.*?)</command-args>", re.DOTALL)
1351def _backfill_skill_events(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1352 """Seed ``skill_events`` from /ll: invocations in user blocks of session JSONL files.
1354 Mirrors :func:`_backfill_messages` but selects ``type == "user"`` records and
1355 matches the ``<command-name>/ll:<name></command-name>`` signal. Populates the
1356 ``skill_events`` table that was added in schema v7 (ENH-1833) but never extended
1357 to include a backfill path (BUG-2283). Used by ``ll-logs stats`` so pre-init
1358 invocations are reflected in skill invocation counts.
1359 """
1360 count = 0
1361 for jsonl_file in jsonl_files:
1362 try:
1363 handle = jsonl_file.open(encoding="utf-8")
1364 except OSError:
1365 continue
1366 with handle:
1367 for line in handle:
1368 line = line.strip()
1369 if not line:
1370 continue
1371 try:
1372 record = json.loads(line)
1373 except json.JSONDecodeError:
1374 continue
1375 if record.get("type") != "user":
1376 continue
1377 session_id = record.get("sessionId")
1378 ts = str(record.get("timestamp") or "")
1379 content = record.get("message", {}).get("content", "")
1380 if isinstance(content, list):
1381 text = ""
1382 for block in content:
1383 if isinstance(block, dict):
1384 text = block.get("text", "")
1385 if text:
1386 break
1387 elif isinstance(content, str):
1388 text = content
1389 else:
1390 text = ""
1391 if not text:
1392 continue
1393 m = _BACKFILL_SKILL_RE.search(text)
1394 if not m:
1395 continue
1396 skill_name = m.group(1)
1397 if skill_name.endswith("</command-name>"):
1398 skill_name = skill_name[: -len("</command-name>")]
1399 args_m = _BACKFILL_ARGS_RE.search(text)
1400 args = args_m.group(1).strip()[:200] if args_m else ""
1401 conn.execute(
1402 "INSERT INTO skill_events(ts, session_id, skill_name, args) VALUES(?, ?, ?, ?)",
1403 (ts, str(session_id) if session_id else None, skill_name, args),
1404 )
1405 _index(
1406 conn,
1407 content=skill_name,
1408 kind="skill",
1409 ref=str(session_id) if session_id else "",
1410 anchor=str(jsonl_file),
1411 ts=ts,
1412 )
1413 count += 1
1414 return count
1417def mine_corrections_from_messages(conn: sqlite3.Connection, config: dict | None = None) -> int:
1418 """Scan ``message_events`` and insert matching rows into ``user_corrections``.
1420 Designed for both the one-time retroactive pass over existing rows and
1421 repeated calls during backfill; idempotent via ``INSERT OR IGNORE`` +
1422 ``idx_corrections_dedup``. Only writes a ``search_index`` entry when the
1423 row is actually inserted (rowcount == 1) to avoid duplicate FTS rows.
1424 Gated by ``analytics.capture.corrections`` (ENH-1841).
1426 Returns the count of newly inserted correction rows.
1427 """
1428 extra_patterns: list[str] = []
1429 if config is not None:
1430 from little_loops.config.features import AnalyticsCaptureConfig
1432 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
1433 if not capture.corrections:
1434 return 0
1435 extra_patterns = capture.correction_patterns
1437 count = 0
1438 rows = conn.execute("SELECT ts, session_id, content FROM message_events").fetchall()
1439 for ts, session_id, content in rows:
1440 if not content or not is_correction(content, extra_patterns=extra_patterns):
1441 continue
1442 text = content[:512]
1443 cursor = conn.execute(
1444 "INSERT OR IGNORE INTO user_corrections(ts, session_id, content, source)"
1445 " VALUES(?, ?, ?, 'backfill')",
1446 (ts, session_id, text),
1447 )
1448 if cursor.rowcount:
1449 _index(
1450 conn,
1451 content=text,
1452 kind="correction",
1453 ref=session_id or "",
1454 anchor="backfill",
1455 ts=ts,
1456 )
1457 count += 1
1458 return count
1461# ---------------------------------------------------------------------------
1462# Compaction — LCM-style summary DAG (FEAT-1712)
1463# ---------------------------------------------------------------------------
1466def _estimate_tokens(text: str) -> int:
1467 """Rough token estimate using the LCM convention: 4 characters per token."""
1468 return len(text) // 4
1471def _summarize_block(
1472 messages: list[str],
1473 budget: int,
1474 *,
1475 model: str | None = None,
1476 timeout: int = 60,
1477) -> str:
1478 """Summarize block_text to fit within budget tokens, with convergence guarantee.
1480 LCM Algorithm 3 three-level escalation:
1482 1. **Level 1**: Normal LLM summary (preserve details), target = budget.
1483 Accepted only if ``_estimate_tokens(result) < _estimate_tokens(input)``.
1484 2. **Level 2**: Aggressive bullet-point LLM summary at ``budget // 2``.
1485 Triggered when level-1 output is not smaller than input.
1486 3. **Level 3**: Deterministic truncation — ``min(budget * 4, 2048)`` characters.
1487 Guaranteed to produce output ≤ input by construction.
1488 Escalations are logged at WARNING level for operator visibility.
1489 """
1491 block_text = "\n---\n".join(messages)
1493 est_input = _estimate_tokens(block_text)
1495 # Short-circuit: for very small inputs an LLM summary cannot be meaningfully
1496 # smaller than the input — skip directly to deterministic truncation.
1497 if est_input < 25:
1498 return block_text[: min(budget * 4, 2048)]
1500 # -- Level 1: normal prose summary -------------------------------------------------
1501 level1_prompt = (
1502 "Summarize these session messages concisely (2-3 paragraphs), capturing key "
1503 "topics, decisions, and outcomes. Target approximately "
1504 f"{budget} tokens:\n\n" + block_text
1505 )
1506 result = _call_llm_for_summary(level1_prompt, model=model, timeout=timeout)
1507 if result and _estimate_tokens(result) < est_input:
1508 return result
1510 # -- Level 2: aggressive bullet-point summary at half budget -----------------------
1511 if result:
1512 logger.warning(
1513 "_summarize_block: level-1 summary not smaller than input "
1514 "(est_output=%d >= est_input=%d); escalating to level 2",
1515 _estimate_tokens(result),
1516 est_input,
1517 )
1518 else:
1519 logger.warning("_summarize_block: level-1 LLM call failed; escalating to level 2")
1520 level2_budget = max(budget // 2, 64)
1521 level2_prompt = (
1522 "Summarize these session messages as a compact bullet list. Be extremely terse: "
1523 "one line per key point, no preamble or commentary. Target approximately "
1524 f"{level2_budget} tokens:\n\n" + block_text
1525 )
1526 result = _call_llm_for_summary(level2_prompt, model=model, timeout=timeout)
1527 if result and _estimate_tokens(result) < est_input:
1528 return result
1530 # -- Level 3: deterministic truncation (guaranteed convergence) --------------------
1531 if result:
1532 logger.warning(
1533 "_summarize_block: level-2 summary not smaller than input "
1534 "(est_output=%d >= est_input=%d); escalating to level 3",
1535 _estimate_tokens(result),
1536 est_input,
1537 )
1538 else:
1539 logger.warning("_summarize_block: level-2 LLM call failed; escalating to level 3")
1540 # Truncation: min(budget * 4, 2048) chars. The 2048 cap (~512 tokens at 4 chars/token)
1541 # follows the LCM paper's level-3 constant, providing a strict convergence guarantee.
1542 max_chars = min(budget * 4, 2048)
1543 return block_text[:max_chars]
1546def _call_llm_for_summary(
1547 prompt: str,
1548 *,
1549 model: str | None = None,
1550 timeout: int = 60,
1551) -> str | None:
1552 """Call the host LLM for a summary and extract the prose ``result`` field.
1554 Returns the extracted prose string on success, or ``None`` if the LLM call
1555 failed or produced an unparseable response (allowing escalation logic to
1556 fall through to the next level).
1557 """
1559 try:
1560 inv = resolve_host().build_blocking_json(prompt=prompt, model=model)
1561 proc = subprocess.run(
1562 [inv.binary, *inv.args],
1563 env={**os.environ, **inv.env},
1564 capture_output=True,
1565 text=True,
1566 timeout=timeout,
1567 )
1568 except subprocess.TimeoutExpired:
1569 logger.warning("_call_llm_for_summary: LLM call timed out after %ds", timeout)
1570 return None
1571 except FileNotFoundError:
1572 logger.error(
1573 "_call_llm_for_summary: %s CLI not found. Install the active host CLI "
1574 "(see LL_HOST_CLI).",
1575 inv.binary,
1576 )
1577 return None
1579 if proc.returncode != 0:
1580 stderr_preview = proc.stderr.strip()[:200] if proc.stderr else "(no stderr)"
1581 logger.error(
1582 "_call_llm_for_summary: %s CLI returned exit code %d (stderr: %s)",
1583 inv.binary,
1584 proc.returncode,
1585 stderr_preview,
1586 )
1587 return None
1589 if not proc.stdout.strip():
1590 stderr_info = proc.stderr.strip()[:200] if proc.stderr else ""
1591 logger.error(
1592 "_call_llm_for_summary: %s CLI returned empty stdout on exit 0"
1593 + (f" (stderr: {stderr_info})" if stderr_info else "")
1594 )
1595 return None
1597 # Parse the JSON envelope and extract the 'result' field — see
1598 # evaluate_llm_structured() at fsm/evaluators.py:832-880 for the
1599 # canonical envelope-parsing pattern.
1600 try:
1601 stdout = proc.stdout.strip()
1602 try:
1603 envelope = json.loads(stdout)
1604 except json.JSONDecodeError:
1605 # Try JSONL: take the last non-empty line
1606 lines = [line for line in stdout.split("\n") if line.strip()]
1607 if not lines:
1608 raise
1609 envelope = json.loads(lines[-1])
1611 # Check for structured-output retry exhaustion or legacy is_error
1612 if envelope.get("subtype") == "error_max_structured_output_retries":
1613 logger.error(
1614 "_call_llm_for_summary: %s CLI could not produce valid output after retries",
1615 inv.binary,
1616 )
1617 return None
1618 if envelope.get("is_error", False):
1619 err_text = str(envelope.get("result", "") or "")[:200]
1620 logger.error(
1621 "_call_llm_for_summary: %s CLI reported error: %s",
1622 inv.binary,
1623 err_text,
1624 )
1625 return None
1627 # Extract the result field (plain prose; no --json-schema here)
1628 result = envelope.get("result", "")
1629 if not result:
1630 logger.error(
1631 "_call_llm_for_summary: empty result field in %s CLI response",
1632 inv.binary,
1633 )
1634 return None
1635 return str(result)
1637 except (json.JSONDecodeError, TypeError, ValueError) as e:
1638 raw_preview = proc.stdout[:300] if proc.stdout else "(empty)"
1639 logger.error(
1640 "_call_llm_for_summary: failed to parse LLM response: %s (raw: %s)",
1641 e,
1642 raw_preview,
1643 )
1644 return None
1647def _compact_session_conn(
1648 conn: sqlite3.Connection,
1649 session_id: str,
1650 budget: int = 4096,
1651 *,
1652 model: str | None = None,
1653 timeout: int = 60,
1654) -> int:
1655 """Compact one session using an existing connection. Returns new leaf node count.
1657 Greedy single-pass block grouping: token estimate ``len(s) // 4``. Each block
1658 gets one ``leaf`` summary_node; if the session accumulates ≥ 2 leaves a single
1659 ``condensed`` node is inserted (or silently skipped if one already exists via
1660 ``INSERT OR IGNORE`` + ``idx_summary_nodes_condensed_dedup``). Leaf dedup is
1661 handled by ``idx_summary_nodes_leaf_dedup`` on ``(session_id, ts_start, ts_end)``.
1662 """
1663 rows = conn.execute(
1664 "SELECT id, ts, content FROM message_events WHERE session_id = ? ORDER BY ts, id",
1665 (session_id,),
1666 ).fetchall()
1668 if not rows:
1669 return 0
1671 # Greedy block accumulation
1672 blocks: list[list[tuple[int, str, str]]] = []
1673 current: list[tuple[int, str, str]] = []
1674 current_tokens = 0
1676 for row in rows:
1677 msg_id, ts, content = row[0], row[1], row[2] or ""
1678 tok = _estimate_tokens(content)
1679 if current_tokens + tok > budget and current:
1680 blocks.append(current)
1681 current = [(msg_id, ts, content)]
1682 current_tokens = tok
1683 else:
1684 current.append((msg_id, ts, content))
1685 current_tokens += tok
1686 if current:
1687 blocks.append(current)
1689 now = _now()
1690 new_leaves = 0
1692 for block in blocks:
1693 ts_start = block[0][1]
1694 ts_end = block[-1][1]
1695 msg_ids = [r[0] for r in block]
1696 contents = [r[2] for r in block]
1698 summary = _summarize_block(contents, budget, model=model, timeout=timeout)
1699 cursor = conn.execute(
1700 "INSERT OR IGNORE INTO summary_nodes"
1701 "(kind, content, tokens, session_id, ts_start, ts_end, created_at)"
1702 " VALUES('leaf', ?, ?, ?, ?, ?, ?)",
1703 (summary, _estimate_tokens(summary), session_id, ts_start, ts_end, now),
1704 )
1705 if cursor.rowcount:
1706 leaf_id = cursor.lastrowid
1707 conn.executemany(
1708 "INSERT OR IGNORE INTO summary_spans(summary_id, message_event_id) VALUES(?, ?)",
1709 [(leaf_id, mid) for mid in msg_ids],
1710 )
1711 new_leaves += 1
1713 # Condensed node: one per session, summarises all leaves.
1714 all_leaves = conn.execute(
1715 "SELECT id, content FROM summary_nodes"
1716 " WHERE kind='leaf' AND session_id=? ORDER BY ts_start",
1717 (session_id,),
1718 ).fetchall()
1720 if len(all_leaves) >= 2:
1721 leaf_summaries = [r[1] for r in all_leaves]
1722 condensed_text = _summarize_block(leaf_summaries, budget, model=model, timeout=timeout)
1723 cursor = conn.execute(
1724 "INSERT OR IGNORE INTO summary_nodes"
1725 "(kind, content, tokens, session_id, ts_start, ts_end, created_at)"
1726 " VALUES('condensed', ?, ?, ?, NULL, NULL, ?)",
1727 (condensed_text, _estimate_tokens(condensed_text), session_id, now),
1728 )
1729 if cursor.rowcount:
1730 condensed_id = cursor.lastrowid
1731 conn.execute(
1732 "UPDATE summary_nodes SET parent_id = ?"
1733 " WHERE session_id = ? AND kind = 'leaf' AND parent_id IS NULL",
1734 (condensed_id, session_id),
1735 )
1737 return new_leaves
1740def _compact_sessions(
1741 conn: sqlite3.Connection,
1742 config: dict | None = None,
1743 max_sessions: int | None = None,
1744) -> int:
1745 """Compact all sessions in the sessions table; returns total new leaf nodes created.
1747 Gated by ``history.compaction.enabled`` (default ``false``). Skips silently when
1748 disabled so backfill() callers that omit config are unaffected.
1750 When ``cross_session_enabled`` is True (default), runs a recursive cross-session
1751 condensation pass after per-session compaction: existing condensed nodes are
1752 grouped level-by-level by token budget, summarised, and inserted as higher-order
1753 condensed nodes (``session_id=NULL``, ``level=1+``) until exactly one project-root
1754 summary node remains (ENH-1954).
1756 Args:
1757 max_sessions: When set, caps the number of sessions compacted in this run
1758 (useful for incremental first-time backfills on large databases).
1759 """
1760 from little_loops.config.features import CompactionConfig
1762 raw = config.get("history", {}).get("compaction", {}) if config else {}
1763 compact_cfg = CompactionConfig.from_dict(raw)
1764 if not compact_cfg.enabled:
1765 return 0
1767 rows = conn.execute("SELECT session_id FROM sessions ORDER BY started_at DESC").fetchall()
1768 if max_sessions is not None:
1769 rows = rows[:max_sessions]
1770 total = 0
1771 for row in rows:
1772 total += _compact_session_conn(
1773 conn,
1774 row[0],
1775 budget=compact_cfg.budget_tokens,
1776 model=compact_cfg.model,
1777 timeout=compact_cfg.timeout,
1778 )
1780 # -- Cross-session condensation (ENH-1954) ---------------------------------
1781 if not compact_cfg.cross_session_enabled:
1782 return total
1784 now = _now()
1785 level = 1
1786 max_level = compact_cfg.max_level # None = unlimited
1788 while True:
1789 # Collect condensed nodes at the current level.
1790 # Level 0 = per-session condensed; level 1+ = cross-session.
1791 condensed = conn.execute(
1792 "SELECT id, content, tokens, session_id FROM summary_nodes"
1793 " WHERE kind='condensed' AND level = ?"
1794 " ORDER BY id",
1795 (level - 1,),
1796 ).fetchall()
1798 if len(condensed) <= 1:
1799 break # nothing to roll up, or already at root
1801 # Group by token budget — same greedy algorithm as _compact_session_conn
1802 groups: list[list[tuple[int, str, int, str | None]]] = []
1803 current: list[tuple[int, str, int, str | None]] = []
1804 current_tokens = 0
1806 for row in condensed:
1807 node_id, content, tokens, sess_id = (
1808 row[0],
1809 row[1],
1810 row[2] or 0,
1811 row[3],
1812 )
1813 if current_tokens + tokens > compact_cfg.budget_tokens and current:
1814 groups.append(current)
1815 current = [(node_id, content, tokens, sess_id)]
1816 current_tokens = tokens
1817 else:
1818 current.append((node_id, content, tokens, sess_id))
1819 current_tokens += tokens
1820 if current:
1821 groups.append(current)
1823 for group in groups:
1824 member_ids = [g[0] for g in group]
1825 contents = [g[1] for g in group]
1827 summary = _summarize_block(
1828 contents,
1829 compact_cfg.budget_tokens,
1830 model=compact_cfg.model,
1831 timeout=compact_cfg.timeout,
1832 )
1834 # Compute ts_start/ts_end for the dedup index.
1835 # Level-1 members are per-session condensed nodes (session_id NOT NULL
1836 # but ts_start=NULL). Query leaf descendants via session_id to get
1837 # real timestamps. Level-2+ members already have ts_start/ts_end set.
1838 if level == 1:
1839 sess_ids = [g[3] for g in group if g[3] is not None]
1840 if sess_ids:
1841 ph = ",".join(["?"] * len(sess_ids))
1842 ts_row = conn.execute(
1843 f"SELECT MIN(ts_start), MAX(ts_end) FROM summary_nodes"
1844 f" WHERE kind='leaf' AND session_id IN ({ph})",
1845 sess_ids,
1846 ).fetchone()
1847 ts_start = ts_row[0] if ts_row else None
1848 ts_end = ts_row[1] if ts_row else None
1849 else:
1850 ts_start, ts_end = None, None
1851 else:
1852 ph = ",".join(["?"] * len(member_ids))
1853 ts_row = conn.execute(
1854 f"SELECT MIN(ts_start), MAX(ts_end) FROM summary_nodes WHERE id IN ({ph})",
1855 member_ids,
1856 ).fetchone()
1857 ts_start = ts_row[0] if ts_row else None
1858 ts_end = ts_row[1] if ts_row else None
1860 cursor = conn.execute(
1861 "INSERT OR IGNORE INTO summary_nodes"
1862 "(kind, content, tokens, session_id, ts_start, ts_end, created_at, level)"
1863 " VALUES('condensed', ?, ?, NULL, ?, ?, ?, ?)",
1864 (summary, _estimate_tokens(summary), ts_start, ts_end, now, level),
1865 )
1866 if cursor.rowcount:
1867 parent_id: int | None = cursor.lastrowid
1868 else:
1869 # Node already exists (idempotent re-run) — look up its id
1870 existing = conn.execute(
1871 "SELECT id FROM summary_nodes"
1872 " WHERE kind='condensed' AND session_id IS NULL"
1873 " AND level = ? AND ts_start = ? AND ts_end = ?",
1874 (level, ts_start, ts_end),
1875 ).fetchone()
1876 parent_id = existing[0] if existing else None
1878 if parent_id is not None:
1879 ph = ",".join(["?"] * len(member_ids))
1880 conn.execute(
1881 f"UPDATE summary_nodes SET parent_id = ?"
1882 f" WHERE id IN ({ph}) AND parent_id IS NULL",
1883 [parent_id] + member_ids,
1884 )
1886 # Depth-limit check
1887 if max_level is not None and level >= max_level:
1888 break
1890 level += 1
1892 return total
1895def compact_session(
1896 session_id: str,
1897 db: Path | str = DEFAULT_DB_PATH,
1898 *,
1899 config: dict | None = None,
1900) -> int:
1901 """Summarize message_events for one session into summary_nodes and summary_spans.
1903 Idempotent: repeated calls do not create duplicate nodes (INSERT OR IGNORE +
1904 partial unique indexes). Uses LCM Algorithm 3 three-level escalation (level 1:
1905 normal LLM summary → level 2: aggressive bullet-point LLM summary → level 3:
1906 deterministic truncation) so a leaf node is always produced. Returns the count
1907 of new leaf nodes created.
1908 """
1909 from little_loops.config.features import CompactionConfig
1911 raw = config.get("history", {}).get("compaction", {}) if config else {}
1912 compact_cfg = CompactionConfig.from_dict(raw)
1913 conn = connect(db)
1914 try:
1915 result = _compact_session_conn(
1916 conn,
1917 session_id,
1918 budget=compact_cfg.budget_tokens,
1919 model=compact_cfg.model,
1920 timeout=compact_cfg.timeout,
1921 )
1922 conn.commit()
1923 finally:
1924 conn.close()
1925 return result
1928def _backfill_sessions(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1929 """Seed ``sessions`` table by mapping each JSONL file to its session_id.
1931 Reads just enough of each file to extract the first ``sessionId`` value,
1932 then inserts one row per unique session. ``INSERT OR IGNORE`` + PRIMARY KEY
1933 makes repeated calls idempotent (ENH-1710).
1934 """
1935 count = 0
1936 for jsonl_file in jsonl_files:
1937 try:
1938 handle = jsonl_file.open(encoding="utf-8")
1939 except OSError:
1940 continue
1941 with handle:
1942 for line in handle:
1943 line = line.strip()
1944 if not line:
1945 continue
1946 try:
1947 record = json.loads(line)
1948 except json.JSONDecodeError:
1949 continue
1950 session_id = record.get("sessionId")
1951 if session_id:
1952 cur = conn.execute(
1953 "INSERT OR IGNORE INTO sessions(session_id, jsonl_path) VALUES(?, ?)",
1954 (str(session_id), str(jsonl_file)),
1955 )
1956 count += cur.rowcount
1957 break # one session_id per file is sufficient
1958 return count
1961def _mtime(path: Path) -> float:
1962 """Return file modification time as a Unix float, or 0.0 if inaccessible."""
1963 try:
1964 return path.stat().st_mtime
1965 except OSError:
1966 return 0.0
1969def backfill_snapshots(
1970 db: Path | str = DEFAULT_DB_PATH,
1971 *,
1972 issues_dir: Path | None = None,
1973) -> int:
1974 """Hydrate ``issue_snapshots`` from all ``.md`` files under *issues_dir*.
1976 Idempotent via ``INSERT OR IGNORE`` on the ``(issue_id, transition)`` dedup
1977 index. Also indexes each snapshot in ``search_index`` with ``kind="snapshot"``.
1978 Returns the number of rows inserted (0 when *issues_dir* is absent or empty).
1979 """
1980 issues_dir = issues_dir if issues_dir is not None else Path(".issues")
1981 if not issues_dir.is_dir():
1982 return 0
1983 conn = connect(db)
1984 try:
1985 count = _backfill_snapshots(conn, issues_dir)
1986 conn.commit()
1987 finally:
1988 conn.close()
1989 return count
1992def backfill(
1993 db: Path | str = DEFAULT_DB_PATH,
1994 *,
1995 issues_dir: Path | None = None,
1996 loops_dir: Path | None = None,
1997 jsonl_files: list[Path] | None = None,
1998 config: dict | None = None,
1999 max_sessions: int | None = None,
2000) -> dict[str, int]:
2001 """Populate the database from existing on-disk sources.
2003 Reads issue-file frontmatter, FSM loop-state JSON, and (optionally) session
2004 JSONL tool-use blocks plus user-message blocks. Returns a per-kind count of
2005 rows inserted. Sources that are absent are skipped silently.
2006 """
2007 issues_dir = issues_dir if issues_dir is not None else Path(".issues")
2008 loops_dir = loops_dir if loops_dir is not None else Path(".loops")
2009 conn = connect(db)
2010 counts: dict[str, int] = {
2011 "issues": 0,
2012 "loops": 0,
2013 "tools": 0,
2014 "messages": 0,
2015 "assistant_messages": 0,
2016 "sessions": 0,
2017 "corrections": 0,
2018 "summaries": 0,
2019 "snapshots": 0,
2020 "skill_events": 0,
2021 }
2022 try:
2023 if issues_dir.is_dir():
2024 counts["issues"] = _backfill_issues(conn, issues_dir)
2025 counts["snapshots"] = _backfill_snapshots(conn, issues_dir)
2026 if loops_dir.is_dir():
2027 counts["loops"] = _backfill_loops(conn, loops_dir)
2028 if jsonl_files:
2029 counts["tools"] = _backfill_tool_events(conn, jsonl_files)
2030 counts["messages"] = _backfill_messages(conn, jsonl_files)
2031 counts["assistant_messages"] = _backfill_assistant_messages(conn, jsonl_files)
2032 counts["skill_events"] = _backfill_skill_events(conn, jsonl_files)
2033 counts["sessions"] = _backfill_sessions(conn, jsonl_files)
2034 counts["corrections"] = mine_corrections_from_messages(conn, config)
2035 counts["summaries"] = _compact_sessions(conn, config, max_sessions=max_sessions)
2036 conn.commit()
2037 finally:
2038 conn.close()
2039 return counts
2042def backfill_incremental(
2043 db: Path | str = DEFAULT_DB_PATH,
2044 *,
2045 jsonl_files: list[Path],
2046 since_ts: float | None = None,
2047 config: dict | None = None,
2048) -> dict[str, int]:
2049 """Backfill only JSONL files modified after *since_ts*.
2051 If *since_ts* is ``None``, reads ``last_backfill_ts`` from the ``meta``
2052 table (defaults to 0.0 — all files — when the key is absent or NULL).
2053 On success, writes the current UTC time as the new ``last_backfill_ts``
2054 so the next call automatically skips already-processed files.
2056 Issues and loop-state JSON are NOT backfilled here; this variant is
2057 JSONL-only and designed for low-latency background use in session hooks.
2058 Errors are not suppressed — the caller (session hook) catches them and
2059 logs a warning.
2060 """
2061 conn = connect(db)
2062 counts: dict[str, int] = {
2063 "tools": 0,
2064 "messages": 0,
2065 "assistant_messages": 0,
2066 "skill_events": 0,
2067 "sessions": 0,
2068 "corrections": 0,
2069 }
2070 try:
2071 if since_ts is None:
2072 row = conn.execute("SELECT value FROM meta WHERE key = 'last_backfill_ts'").fetchone()
2073 raw = row[0] if (row and row[0]) else None
2074 if raw:
2075 try:
2076 since_ts = datetime.fromisoformat(str(raw).replace("Z", "+00:00")).timestamp()
2077 except ValueError:
2078 since_ts = 0.0
2079 else:
2080 since_ts = 0.0
2082 filtered = [f for f in jsonl_files if _mtime(f) >= since_ts]
2083 if filtered:
2084 counts["sessions"] = _backfill_sessions(conn, filtered)
2085 counts["tools"] = _backfill_tool_events(conn, filtered)
2086 counts["messages"] = _backfill_messages(conn, filtered)
2088 # Per-table watermark for assistant_messages (added in v11 / ENH-1942).
2089 # When the key is absent the table has never been backfilled (e.g. it
2090 # was added by a schema migration after the last full backfill), so we
2091 # reprocess ALL provided files rather than only the globally-filtered
2092 # subset. This self-heals the historical gap on the first incremental
2093 # run after the migration without requiring a manual full backfill.
2094 _ASST_KEY = "last_backfill_ts_assistant_messages"
2095 _asst_row = conn.execute("SELECT value FROM meta WHERE key = ?", (_ASST_KEY,)).fetchone()
2096 _asst_raw = _asst_row[0] if (_asst_row and _asst_row[0]) else None
2097 if _asst_raw:
2098 try:
2099 _asst_since = datetime.fromisoformat(
2100 str(_asst_raw).replace("Z", "+00:00")
2101 ).timestamp()
2102 except ValueError:
2103 _asst_since = 0.0
2104 else:
2105 _asst_since = 0.0 # never backfilled → process all files
2107 asst_filtered = [f for f in jsonl_files if _mtime(f) >= _asst_since]
2108 if asst_filtered:
2109 counts["assistant_messages"] = _backfill_assistant_messages(conn, asst_filtered)
2110 conn.execute(
2111 "INSERT INTO meta(key, value) VALUES(?, ?) "
2112 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
2113 (_ASST_KEY, _now()),
2114 )
2116 # Per-table watermark for skill_events (added in v7 / ENH-1833, backfill
2117 # path added in BUG-2283). Same self-healing logic as assistant_messages:
2118 # absent key → reprocess all files on the first incremental run after deploy.
2119 _SKILL_KEY = "last_backfill_ts_skill_events"
2120 _skill_row = conn.execute("SELECT value FROM meta WHERE key = ?", (_SKILL_KEY,)).fetchone()
2121 _skill_raw = _skill_row[0] if (_skill_row and _skill_row[0]) else None
2122 if _skill_raw:
2123 try:
2124 _skill_since = datetime.fromisoformat(
2125 str(_skill_raw).replace("Z", "+00:00")
2126 ).timestamp()
2127 except ValueError:
2128 _skill_since = 0.0
2129 else:
2130 _skill_since = 0.0 # never backfilled → process all files
2132 skill_filtered = [f for f in jsonl_files if _mtime(f) >= _skill_since]
2133 if skill_filtered:
2134 counts["skill_events"] = _backfill_skill_events(conn, skill_filtered)
2135 conn.execute(
2136 "INSERT INTO meta(key, value) VALUES(?, ?) "
2137 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
2138 (_SKILL_KEY, _now()),
2139 )
2141 counts["corrections"] = mine_corrections_from_messages(conn, config)
2142 conn.execute(
2143 "INSERT INTO meta(key, value) VALUES('last_backfill_ts', ?) "
2144 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
2145 (_now(),),
2146 )
2147 conn.commit()
2148 finally:
2149 conn.close()
2150 return counts
2153# High-volume tables eligible for age-based pruning.
2154_PRUNABLE_TABLES = ("tool_events", "cli_events", "file_events", "message_events")
2157def prune(
2158 db: Path | str = DEFAULT_DB_PATH,
2159 *,
2160 config: dict | None = None,
2161 dry_run: bool = False,
2162) -> dict:
2163 """Prune raw event rows older than configured max-age, then VACUUM the database.
2165 Both dual gates must be exceeded before any rows are deleted:
2166 - ``min_project_age_days``: project age (MIN(started_at) from sessions table)
2167 - ``min_db_size_mb``: DB file size on disk
2169 High-volume tables pruned: ``tool_events``, ``cli_events``, ``file_events``,
2170 ``message_events``. High-value tables (``issue_events``, ``user_corrections``)
2171 are never pruned.
2173 Args:
2174 db: Path to the history database.
2175 config: Project config dict (reads ``analytics.retention``). ``None`` uses defaults.
2176 dry_run: Count rows that would be deleted without deleting them.
2178 Returns:
2179 dict with keys:
2180 - ``pruned`` (bool): whether pruning ran (gates met and rows eligible)
2181 - ``gate_unmet`` (list[str]): human-readable reason for each unmet gate
2182 - ``project_age_days`` (int): measured project age
2183 - ``db_size_mb`` (float): DB file size in MB
2184 - ``deleted`` (dict[str, int]): row counts per table (actual or projected)
2185 - ``vacuumed`` (bool): whether VACUUM ran (always False in dry_run)
2186 """
2187 from little_loops.config.features import RetentionConfig
2189 raw = (config or {}).get("analytics", {}).get("retention", {})
2190 retention_cfg = RetentionConfig.from_dict(raw)
2192 db_path = Path(db)
2193 result: dict = {
2194 "pruned": False,
2195 "gate_unmet": [],
2196 "project_age_days": 0,
2197 "db_size_mb": 0.0,
2198 "deleted": {},
2199 "vacuumed": False,
2200 }
2202 conn = connect(db)
2203 try:
2204 # Gate 1: project age — MIN(started_at) from sessions
2205 row = conn.execute("SELECT MIN(started_at) FROM sessions").fetchone()
2206 oldest_ts = row[0] if row and row[0] else None
2207 if oldest_ts:
2208 try:
2209 oldest_dt = datetime.fromisoformat(oldest_ts.replace("Z", "+00:00"))
2210 project_age_days = (datetime.now(UTC) - oldest_dt).days
2211 except ValueError:
2212 project_age_days = 0
2213 else:
2214 project_age_days = 0
2215 result["project_age_days"] = project_age_days
2217 # Gate 2: DB file size
2218 db_size_mb = db_path.stat().st_size / (1024 * 1024) if db_path.exists() else 0.0
2219 result["db_size_mb"] = round(db_size_mb, 2)
2221 # Evaluate gates
2222 gates_unmet: list[str] = []
2223 if project_age_days < retention_cfg.min_project_age_days:
2224 gates_unmet.append(
2225 f"project age {project_age_days}d < {retention_cfg.min_project_age_days}d"
2226 )
2227 if db_size_mb < retention_cfg.min_db_size_mb:
2228 gates_unmet.append(f"db size {db_size_mb:.1f}MB < {retention_cfg.min_db_size_mb}MB")
2229 result["gate_unmet"] = gates_unmet
2231 if gates_unmet:
2232 return result
2234 if retention_cfg.raw_event_max_age_days is None:
2235 result["pruned"] = True
2236 return result
2238 cutoff = datetime.now(UTC) - timedelta(days=retention_cfg.raw_event_max_age_days)
2239 cutoff_str = cutoff.strftime("%Y-%m-%dT%H:%M:%SZ")
2241 deleted: dict[str, int] = {}
2242 for table in _PRUNABLE_TABLES:
2243 count_row = conn.execute(
2244 f"SELECT COUNT(*) FROM {table} WHERE ts < ?", (cutoff_str,)
2245 ).fetchone()
2246 deleted[table] = count_row[0] if count_row else 0
2247 if not dry_run and deleted[table] > 0:
2248 conn.execute(f"DELETE FROM {table} WHERE ts < ?", (cutoff_str,))
2250 result["deleted"] = deleted
2251 result["pruned"] = True
2253 if not dry_run:
2254 conn.commit()
2255 finally:
2256 conn.close()
2258 # VACUUM outside the original connection to avoid transaction conflicts
2259 if result["pruned"] and not dry_run:
2260 try:
2261 vac_conn = sqlite3.connect(str(db_path))
2262 _configure_connection(vac_conn)
2263 vac_conn.isolation_level = None
2264 try:
2265 vac_conn.execute("VACUUM")
2266 result["vacuumed"] = True
2267 finally:
2268 vac_conn.close()
2269 except sqlite3.Error as exc:
2270 logger.warning("prune: VACUUM failed: %s", exc)
2272 return result
2275# ---------------------------------------------------------------------------
2276# Correction retirement (ENH-2046)
2277# ---------------------------------------------------------------------------
2280def record_retirement(
2281 db: Path | str = DEFAULT_DB_PATH,
2282 topic_fingerprint: str = "",
2283 rule_id: str = "",
2284 session_id: str = "",
2285) -> None:
2286 """Mark a recurring-correction cluster as addressed.
2288 Uses INSERT OR REPLACE so a second call for the same fingerprint updates
2289 the record rather than duplicating it. ``rule_id`` should be the
2290 ``decisions.yaml`` entry ID (e.g. ``BEHAVIOR-001``) or ``"claude-md"``
2291 when the rule was written directly into CLAUDE.md.
2292 """
2293 if not topic_fingerprint:
2294 return
2295 conn = connect(db)
2296 try:
2297 conn.execute(
2298 "INSERT OR REPLACE INTO correction_retirements"
2299 "(topic_fingerprint, rule_id, addressed_at, session_id) VALUES (?, ?, ?, ?)",
2300 (topic_fingerprint, rule_id or None, _now(), session_id or None),
2301 )
2302 conn.commit()
2303 finally:
2304 conn.close()
2307def list_retirements(
2308 db: Path | str = DEFAULT_DB_PATH,
2309) -> list[dict]:
2310 """Return all correction retirement records, newest first.
2312 Returns an empty list when the DB does not exist or the
2313 ``correction_retirements`` table has not yet been created.
2314 """
2315 db_path = Path(db)
2316 if not db_path.exists():
2317 return []
2318 conn = connect(db)
2319 try:
2320 rows = conn.execute(
2321 "SELECT topic_fingerprint, rule_id, addressed_at, session_id"
2322 " FROM correction_retirements ORDER BY addressed_at DESC"
2323 ).fetchall()
2324 return [dict(r) for r in rows]
2325 except sqlite3.OperationalError:
2326 return []
2327 finally:
2328 conn.close()
2331# ---------------------------------------------------------------------------
2332# JSONL export (for visualization / external tooling)
2333# ---------------------------------------------------------------------------
2335# Maps the public type name used in exported records to (table, timestamp_column).
2336_EXPORT_TABLE_MAP: dict[str, tuple[str, str]] = {
2337 "session": ("sessions", "started_at"),
2338 "issue_event": ("issue_events", "ts"),
2339 "issue_snapshot": ("issue_snapshots", "ts"),
2340 "skill_event": ("skill_events", "ts"),
2341 "loop_event": ("loop_events", "ts"),
2342 "correction": ("user_corrections", "ts"),
2343 "summary_node": ("summary_nodes", "created_at"),
2344 "message_event": ("message_events", "ts"),
2345}
2347_EXPORT_DEFAULT_TABLES = [
2348 "session",
2349 "issue_event",
2350 "issue_snapshot",
2351 "skill_event",
2352 "loop_event",
2353 "correction",
2354 "summary_node",
2355]
2358def export_history(
2359 db: Path | str = DEFAULT_DB_PATH,
2360 *,
2361 tables: list[str] | None = None,
2362 since: str | None = None,
2363 include_messages: bool = False,
2364) -> Generator[dict, None, None]:
2365 """Yield rows from history.db as dicts with a ``type`` key (JSONL export).
2367 Each yielded dict has a ``"type"`` field identifying the source table so
2368 records from multiple tables can be mixed in a single stream and later
2369 distinguished by a visualizer.
2371 Args:
2372 db: Path to the history database (default: ``.ll/history.db``).
2373 tables: Type names to include. Defaults to all non-message tables.
2374 Valid values: ``session``, ``issue_event``, ``issue_snapshot``,
2375 ``skill_event``, ``loop_event``, ``correction``, ``summary_node``,
2376 ``message_event``.
2377 since: ISO 8601 datetime string; only rows at or after this timestamp are
2378 returned, filtered per-table using the relevant timestamp column.
2379 include_messages: When ``True`` and *tables* is not given, also include
2380 ``message_events`` (~46 K rows by default). Ignored when *tables*
2381 is specified explicitly.
2382 """
2383 db_path = Path(db)
2384 if not db_path.exists():
2385 return
2387 if tables is None:
2388 selected = list(_EXPORT_DEFAULT_TABLES)
2389 if include_messages:
2390 selected.append("message_event")
2391 else:
2392 selected = list(tables)
2394 conn = connect(db_path)
2395 try:
2396 for type_name in selected:
2397 entry = _EXPORT_TABLE_MAP.get(type_name)
2398 if entry is None:
2399 logger.warning("export_history: unknown type %r — skipped", type_name)
2400 continue
2401 table, ts_col = entry
2402 try:
2403 if since:
2404 cursor = conn.execute(
2405 f"SELECT * FROM {table} WHERE {ts_col} >= ? ORDER BY {ts_col}",
2406 (since,),
2407 )
2408 else:
2409 cursor = conn.execute(f"SELECT * FROM {table} ORDER BY {ts_col}")
2410 for row in cursor:
2411 d = dict(row)
2412 d["type"] = type_name
2413 yield d
2414 except sqlite3.OperationalError as exc:
2415 logger.warning("export_history: skipping %s: %s", table, exc)
2416 finally:
2417 conn.close()