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