Coverage for little_loops / session_store.py: 12%
554 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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 search(db,...): FTS5 full-text query with BM25 ranking
29 recent(db,...): recent rows for a given event kind
30 is_correction(text): return True if text matches a user-correction signal
31 record_correction(db,...): write one row to ``user_corrections`` + search_index
32 record_skill_event(db,...): write one row to ``skill_events`` + search_index
33 cli_event_context(db,...): context manager: INSERT on enter, UPDATE exit_code+duration on exit
34"""
36from __future__ import annotations
38import hashlib
39import json
40import logging
41import re
42import sqlite3
43import subprocess
44import threading
45import time
46from collections.abc import Generator, Sequence
47from contextlib import contextmanager
48from datetime import UTC, datetime
49from pathlib import Path
50from typing import Any
52from little_loops.host_runner import resolve_host
54__all__ = [
55 "DEFAULT_DB_PATH",
56 "SCHEMA_VERSION",
57 "ensure_db",
58 "connect",
59 "SQLiteTransport",
60 "backfill",
61 "backfill_incremental",
62 "mine_corrections_from_messages",
63 "compact_session",
64 "search",
65 "recent",
66 "is_correction",
67 "record_correction",
68 "record_skill_event",
69 "cli_event_context",
70]
72logger = logging.getLogger(__name__)
74DEFAULT_DB_PATH = Path(".ll/history.db")
75SCHEMA_VERSION = 10
77_VALID_KINDS = frozenset({"tool", "file", "issue", "loop", "correction", "message", "skill", "cli"})
78_KIND_TABLE = {
79 "tool": "tool_events",
80 "file": "file_events",
81 "issue": "issue_events",
82 "loop": "loop_events",
83 "correction": "user_corrections",
84 "message": "message_events",
85 "skill": "skill_events",
86 "cli": "cli_events",
87}
89# FSM event types the SQLiteTransport records as loop_events rows.
90_LOOP_EVENT_TYPES = frozenset(
91 {
92 "loop_start",
93 "loop_resume",
94 "loop_complete",
95 "state_enter",
96 "route",
97 "retry_exhausted",
98 "cycle_detected",
99 "max_iterations_summary",
100 }
101)
104_CORRECTION_RE = re.compile(
105 r"^\s*(no[,!]|don'?t\s|stop\s|revert|that'?s\s+wrong|not\s+like\s+that)",
106 re.IGNORECASE,
107)
109_PHRASE_RE = re.compile(
110 r"\b(?:"
111 r"instead"
112 r"|actually\s+(?:that|this|it)\s"
113 r"|you missed"
114 r"|should be\s+(?!fine\b|ok\b|okay\b|good\b|great\b|alright\b|correct\b|right\b)"
115 r"|wrong approach"
116 r"|remember that"
117 r"|always use"
118 r"|never use"
119 r"|from now on"
120 r"|I meant\b.*\bnot\b"
121 r"|not\b.*\buse\b"
122 r")",
123 re.IGNORECASE,
124)
125_REMEMBER_RE = re.compile(r"^!remember\b", re.IGNORECASE)
128def is_correction(text: str, extra_patterns: Sequence[str] = ()) -> bool:
129 """Return True if *text* matches a known user-correction signal.
131 ``extra_patterns`` are raw regex strings appended to the built-in set
132 (``analytics.capture.correction_patterns``). Built-ins always remain active.
133 Invalid patterns are skipped with a warning.
134 """
135 t = text[:512]
136 _extra_re: re.Pattern[str] | None = None
137 if extra_patterns:
138 parts: list[str] = []
139 for p in extra_patterns:
140 try:
141 re.compile(p, re.IGNORECASE)
142 parts.append(f"(?:{p})")
143 except re.error:
144 logger.warning("is_correction: skipping invalid extra_pattern %r", p)
145 if parts:
146 _extra_re = re.compile("|".join(parts), re.IGNORECASE)
147 return bool(
148 _REMEMBER_RE.match(t)
149 or _CORRECTION_RE.match(t)
150 or _PHRASE_RE.search(t)
151 or (_extra_re and _extra_re.search(t))
152 )
155# ---------------------------------------------------------------------------
156# Schema + migrations
157# ---------------------------------------------------------------------------
159# Each entry is the full SQL applied to move the schema from version index to
160# index+1. Migration 0 bootstraps the whole schema; append new entries to
161# evolve it. ``bytes_in`` / ``bytes_out`` / ``cache_hit`` are reserved on
162# ``tool_events`` for FEAT-1160 (Context Window Analytics) so that feature does
163# not require a follow-up migration.
164_MIGRATIONS: list[str] = [
165 """
166 CREATE TABLE tool_events (
167 id INTEGER PRIMARY KEY AUTOINCREMENT,
168 ts TEXT NOT NULL,
169 session_id TEXT,
170 tool_name TEXT,
171 args_hash TEXT,
172 result_size INTEGER,
173 bytes_in INTEGER,
174 bytes_out INTEGER,
175 cache_hit INTEGER
176 );
177 CREATE TABLE file_events (
178 id INTEGER PRIMARY KEY AUTOINCREMENT,
179 ts TEXT NOT NULL,
180 session_id TEXT,
181 path TEXT,
182 op TEXT,
183 issue_id TEXT,
184 git_sha TEXT
185 );
186 CREATE TABLE issue_events (
187 id INTEGER PRIMARY KEY AUTOINCREMENT,
188 ts TEXT NOT NULL,
189 issue_id TEXT,
190 transition TEXT,
191 discovered_by TEXT
192 );
193 CREATE TABLE loop_events (
194 id INTEGER PRIMARY KEY AUTOINCREMENT,
195 ts TEXT NOT NULL,
196 loop_name TEXT,
197 state TEXT,
198 transition TEXT,
199 retries INTEGER
200 );
201 CREATE TABLE user_corrections (
202 id INTEGER PRIMARY KEY AUTOINCREMENT,
203 ts TEXT NOT NULL,
204 session_id TEXT,
205 content TEXT,
206 source TEXT
207 );
208 CREATE VIRTUAL TABLE search_index USING fts5(
209 content,
210 kind UNINDEXED,
211 ref UNINDEXED,
212 anchor UNINDEXED,
213 ts UNINDEXED
214 );
215 CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT);
216 """,
217 # v2 (ENH-1621): widen issue_events with completion-summary columns so
218 # ll-history `summary` can be answered from the DB; add message_events for
219 # analyze_workflows() to read user message bodies without re-parsing JSONL.
220 """
221 ALTER TABLE issue_events ADD COLUMN issue_type TEXT;
222 ALTER TABLE issue_events ADD COLUMN priority TEXT;
223 ALTER TABLE issue_events ADD COLUMN completed_date TEXT;
224 ALTER TABLE issue_events ADD COLUMN captured_at TEXT;
225 ALTER TABLE issue_events ADD COLUMN completed_at TEXT;
226 CREATE TABLE message_events (
227 id INTEGER PRIMARY KEY AUTOINCREMENT,
228 ts TEXT NOT NULL,
229 session_id TEXT,
230 content TEXT
231 );
232 """,
233 # v3 (ENH-1690): unique dedup index on issue_events so INSERT OR IGNORE
234 # prevents duplicate rows when backfill() is called multiple times.
235 """
236 CREATE UNIQUE INDEX IF NOT EXISTS idx_issue_events_dedup
237 ON issue_events(issue_id, transition);
238 """,
239 # v4 (ENH-1710): sessions table maps session_id to its JSONL file path,
240 # closing the broken link between event rows and their source log.
241 """
242 CREATE TABLE sessions (
243 session_id TEXT PRIMARY KEY,
244 jsonl_path TEXT NOT NULL,
245 started_at TEXT,
246 project_path TEXT
247 );
248 """,
249 # v5 (ENH-1711): issue_sessions VIEW joins issue_events to message_events via
250 # overlapping timestamps, making the implicit session→issue link explicit and
251 # queryable. Requires captured_at IS NOT NULL; populated by _backfill_issues()
252 # for historical rows and by issue_lifecycle.py emit sites (ENH-1839) for
253 # live-emitted rows.
254 """
255 CREATE VIEW issue_sessions AS
256 SELECT ie.issue_id,
257 me.session_id,
258 s.jsonl_path,
259 MIN(me.ts) AS first_message_ts,
260 MAX(me.ts) AS last_message_ts
261 FROM issue_events ie
262 JOIN message_events me
263 ON me.ts >= ie.captured_at
264 AND (ie.completed_at IS NULL OR me.ts <= ie.completed_at)
265 LEFT JOIN sessions s ON s.session_id = me.session_id
266 WHERE ie.captured_at IS NOT NULL
267 GROUP BY ie.issue_id, me.session_id;
268 """,
269 # v6 (ENH-1830): last_backfill_ts meta key for incremental JSONL backfill at
270 # session start. The meta table already holds arbitrary key/value pairs; this
271 # initialises the sentinel so reads can distinguish "no prior run" (NULL) from
272 # a real ISO 8601 timestamp string.
273 """
274 INSERT OR IGNORE INTO meta(key, value) VALUES('last_backfill_ts', NULL);
275 """,
276 # v7 (ENH-1833): skill_events table records /ll: skill invocations at dispatch
277 # time via the user_prompt_submit hook so ll-session recent --kind skill works.
278 """
279 CREATE TABLE IF NOT EXISTS skill_events (
280 id INTEGER PRIMARY KEY AUTOINCREMENT,
281 ts TEXT NOT NULL,
282 session_id TEXT,
283 skill_name TEXT,
284 args TEXT
285 );
286 """,
287 # v8 (ENH-1848): cli_events table records ll- CLI invocations via cli_event_context()
288 """
289 CREATE TABLE IF NOT EXISTS cli_events (
290 id INTEGER PRIMARY KEY AUTOINCREMENT,
291 ts TEXT NOT NULL,
292 binary TEXT NOT NULL,
293 args TEXT NOT NULL,
294 exit_code INTEGER,
295 duration_ms INTEGER
296 );
297 """,
298 # v9 (ENH-1904): unique dedup index on user_corrections so INSERT OR IGNORE
299 # enforces idempotency during correction mining. Mirrors v3's
300 # idx_issue_events_dedup pattern.
301 """
302 CREATE UNIQUE INDEX IF NOT EXISTS idx_corrections_dedup
303 ON user_corrections(session_id, content);
304 """,
305 # v10 (FEAT-1712): LCM-style hierarchical summary DAG over session history.
306 # summary_nodes holds LLM-generated summaries (via three-level LCM Algorithm 3
307 # escalation: normal → aggressive bullet-point → deterministic truncation) at two
308 # levels: 'leaf' nodes cover a fixed token-budget block of message_events;
309 # 'condensed' nodes summarise a session's leaves. summary_spans links summary
310 # nodes back to the originating message_events rows for lossless drill-down.
311 # FK references are decorative (no PRAGMA foreign_keys; integrity enforced at
312 # the application layer by compact_session's insert ordering + INSERT OR IGNORE).
313 # Partial unique indexes prevent duplicate leaf and condensed nodes per session
314 # across repeated backfill() calls (idempotency via INSERT OR IGNORE).
315 """
316 CREATE TABLE IF NOT EXISTS summary_nodes (
317 id INTEGER PRIMARY KEY AUTOINCREMENT,
318 kind TEXT NOT NULL,
319 content TEXT NOT NULL,
320 tokens INTEGER,
321 parent_id INTEGER REFERENCES summary_nodes(id),
322 session_id TEXT,
323 ts_start TEXT,
324 ts_end TEXT,
325 created_at TEXT NOT NULL
326 );
327 CREATE TABLE IF NOT EXISTS summary_spans (
328 summary_id INTEGER REFERENCES summary_nodes(id),
329 message_event_id INTEGER REFERENCES message_events(id),
330 PRIMARY KEY (summary_id, message_event_id)
331 );
332 CREATE UNIQUE INDEX IF NOT EXISTS idx_summary_nodes_leaf_dedup
333 ON summary_nodes(session_id, ts_start, ts_end) WHERE kind = 'leaf';
334 CREATE UNIQUE INDEX IF NOT EXISTS idx_summary_nodes_condensed_dedup
335 ON summary_nodes(session_id) WHERE kind = 'condensed';
336 CREATE INDEX IF NOT EXISTS idx_summary_nodes_parent_id
337 ON summary_nodes(parent_id);
338 """,
339]
342def _now() -> str:
343 """Return the current UTC time as a Z-suffixed ISO 8601 string."""
344 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
347def _current_version(conn: sqlite3.Connection) -> int:
348 """Return the applied schema version, or 0 if the meta table is absent."""
349 try:
350 row = conn.execute("SELECT value FROM meta WHERE key = 'schema_version'").fetchone()
351 except sqlite3.OperationalError:
352 return 0
353 return int(row[0]) if row else 0
356def _apply_migrations(conn: sqlite3.Connection) -> None:
357 """Apply every migration newer than the database's current version."""
358 version = _current_version(conn)
359 for index in range(version, len(_MIGRATIONS)):
360 conn.executescript(_MIGRATIONS[index])
361 conn.execute(
362 "INSERT INTO meta(key, value) VALUES('schema_version', ?) "
363 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
364 (str(index + 1),),
365 )
366 conn.commit()
369def ensure_db(path: Path | str = DEFAULT_DB_PATH) -> Path:
370 """Create the database at *path* (if needed) and apply pending migrations.
372 Idempotent: safe to call on every session start. The parent directory is
373 created if absent. Returns the resolved database path.
375 On the first call after the ENH-1635 rename, transparently migrates a
376 pre-existing ``.ll/session.db`` (and any ``-shm``/``-wal`` sidecars) to
377 the new ``.ll/history.db`` path. Each sidecar is renamed independently so
378 a single failure does not abort the others; failures are logged at
379 WARNING (the caller in ``hooks/session_start.py`` wraps the whole call
380 in ``contextlib.suppress(Exception)``, which would otherwise silence
381 diagnostic context).
382 """
383 db_path = Path(path)
384 legacy = db_path.parent / "session.db"
385 if legacy.exists() and not db_path.exists():
386 for suffix in ("", "-shm", "-wal"):
387 src = legacy.parent / f"session.db{suffix}"
388 if src.exists():
389 dst = db_path.parent / f"history.db{suffix}"
390 try:
391 src.rename(dst)
392 logger.info("session_store: migrated %s -> %s", src, dst)
393 except OSError:
394 logger.warning(
395 "session_store: legacy rename failed for %s; continuing with fresh db",
396 src,
397 exc_info=True,
398 )
399 break
400 db_path.parent.mkdir(parents=True, exist_ok=True)
401 conn = sqlite3.connect(str(db_path))
402 try:
403 _apply_migrations(conn)
404 finally:
405 conn.close()
406 return db_path
409def connect(path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
410 """Open a connection to the session database, ensuring the schema first.
412 Rows are returned as :class:`sqlite3.Row` so callers can index by name.
413 """
414 db_path = ensure_db(path)
415 conn = sqlite3.connect(str(db_path))
416 conn.row_factory = sqlite3.Row
417 return conn
420def _index(
421 conn: sqlite3.Connection,
422 *,
423 content: str,
424 kind: str,
425 ref: str,
426 anchor: str,
427 ts: str,
428) -> None:
429 """Insert one row into the FTS5 ``search_index`` table."""
430 conn.execute(
431 "INSERT INTO search_index(content, kind, ref, anchor, ts) VALUES(?, ?, ?, ?, ?)",
432 (content, kind, ref, anchor, ts),
433 )
436def write_file_event(
437 db_path: Path | str,
438 session_id: str | None,
439 path: str,
440 op: str,
441 issue_id: str | None = None,
442 config: dict | None = None,
443) -> None:
444 """Write one row to ``file_events`` and index it in ``search_index``.
446 Gated by ``analytics.capture.file_events`` (ENH-1841): when ``config`` is
447 provided and the flag is ``false``, the write is suppressed. Missing ``capture``
448 key defaults to permissive (no behavior change).
449 """
450 if config is not None:
451 from little_loops.config.features import AnalyticsCaptureConfig
453 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
454 if not capture.file_events:
455 return
456 conn = connect(db_path)
457 ts = _now()
458 try:
459 conn.execute(
460 "INSERT INTO file_events(ts, session_id, path, op, issue_id, git_sha) "
461 "VALUES(?, ?, ?, ?, ?, ?)",
462 (ts, session_id, path, op, issue_id, None),
463 )
464 _index(conn, content=path, kind="file", ref=path, anchor=op, ts=ts)
465 conn.commit()
466 finally:
467 conn.close()
470def record_correction(
471 db_path: Path | str,
472 session_id: str | None,
473 content: str,
474 source: str,
475 config: dict | None = None,
476) -> None:
477 """Write one row to ``user_corrections`` and index it in ``search_index``.
479 Gated by ``analytics.capture.corrections`` (ENH-1841): when ``config`` is
480 provided and the flag is ``false``, the write is suppressed. Missing ``capture``
481 key defaults to permissive (no behavior change).
482 """
483 if config is not None:
484 from little_loops.config.features import AnalyticsCaptureConfig
486 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
487 if not capture.corrections:
488 return
489 content = content[:512]
490 conn = connect(db_path)
491 ts = _now()
492 try:
493 conn.execute(
494 "INSERT INTO user_corrections(ts, session_id, content, source) VALUES(?, ?, ?, ?)",
495 (ts, session_id, content, source),
496 )
497 _index(conn, content=content, kind="correction", ref=session_id or "", anchor=source, ts=ts)
498 conn.commit()
499 finally:
500 conn.close()
503def record_skill_event(
504 db_path: Path | str,
505 session_id: str | None,
506 skill_name: str,
507 args: str,
508 config: dict | None = None,
509) -> None:
510 """Write one row to ``skill_events`` and index it in ``search_index``.
512 The ``config`` parameter is a forward-compatibility stub for ENH-1835, which
513 will inject a per-skill analytics gate without changing this signature.
514 """
515 args = args[:200]
516 conn = connect(db_path)
517 ts = _now()
518 try:
519 conn.execute(
520 "INSERT INTO skill_events(ts, session_id, skill_name, args) VALUES(?, ?, ?, ?)",
521 (ts, session_id, skill_name, args),
522 )
523 _index(
524 conn, content=skill_name, kind="skill", ref=session_id or "", anchor=skill_name, ts=ts
525 )
526 conn.commit()
527 finally:
528 conn.close()
531@contextmanager
532def cli_event_context(
533 db_path: Path | str = DEFAULT_DB_PATH,
534 binary: str = "",
535 args: list[str] | None = None,
536 config: dict | None = None,
537) -> Generator[None, None, None]:
538 """Insert a ``cli_events`` row on enter; update exit_code and duration_ms on exit.
540 The ``config`` parameter is a forward-compatibility stub for ENH-1835 gating;
541 it is accepted but not yet used.
542 """
543 if args is None:
544 args = []
545 conn = connect(db_path)
546 start = time.time()
547 ts = _now()
548 cursor = conn.execute(
549 "INSERT INTO cli_events(ts, binary, args) VALUES(?, ?, ?)",
550 (ts, binary, json.dumps(args[:50])),
551 )
552 row_id = cursor.lastrowid
553 conn.commit()
554 exit_code = 0
555 try:
556 yield
557 except BaseException:
558 exit_code = 1
559 raise
560 finally:
561 duration_ms = int((time.time() - start) * 1000)
562 conn.execute(
563 "UPDATE cli_events SET exit_code=?, duration_ms=? WHERE id=?",
564 (exit_code, duration_ms, row_id),
565 )
566 conn.commit()
567 conn.close()
570# ---------------------------------------------------------------------------
571# Query API
572# ---------------------------------------------------------------------------
575def search(
576 db: Path | str = DEFAULT_DB_PATH,
577 *,
578 query: str,
579 limit: int = 20,
580) -> list[dict[str, Any]]:
581 """Run an FTS5 full-text query, returning BM25-ranked results.
583 Each result dict carries ``content``, ``kind``, ``ref``, ``anchor`` (a
584 file:line-style pointer where available), ``ts`` and a numeric ``score``
585 (lower BM25 score = better match).
586 """
587 conn = connect(db)
588 try:
589 rows = conn.execute(
590 "SELECT content, kind, ref, anchor, ts, bm25(search_index) AS score "
591 "FROM search_index WHERE search_index MATCH ? "
592 "ORDER BY score LIMIT ?",
593 (query, limit),
594 ).fetchall()
595 except sqlite3.OperationalError as exc:
596 raise ValueError(f"invalid FTS query {query!r}: {exc}") from exc
597 finally:
598 conn.close()
599 return [dict(row) for row in rows]
602def recent(
603 db: Path | str = DEFAULT_DB_PATH,
604 *,
605 kind: str,
606 limit: int = 20,
607) -> list[dict[str, Any]]:
608 """Return the most recent rows for *kind* (tool, file, issue, loop, correction, message, skill, cli)."""
609 if kind not in _VALID_KINDS:
610 raise ValueError(f"unknown kind {kind!r}; expected one of {sorted(_VALID_KINDS)}")
611 table = _KIND_TABLE[kind]
612 conn = connect(db)
613 try:
614 rows = conn.execute(
615 f"SELECT * FROM {table} ORDER BY id DESC LIMIT ?", # noqa: S608 - table from fixed map
616 (limit,),
617 ).fetchall()
618 finally:
619 conn.close()
620 return [dict(row) for row in rows]
623# ---------------------------------------------------------------------------
624# SQLiteTransport
625# ---------------------------------------------------------------------------
627_ISSUE_TRANSITION_MAP: dict[str, str] = {
628 "issue.completed": "done",
629 "issue.closed": "done",
630 "issue.deferred": "deferred",
631 "issue.skipped": "cancelled",
632 "issue.created": "open",
633 "issue.started": "in_progress",
634}
637def _derive_transition(event_type: str) -> str:
638 """Map an ``issue.*`` event type to the canonical transition/status string."""
639 return _ISSUE_TRANSITION_MAP.get(event_type, event_type.split(".", 1)[1])
642class SQLiteTransport:
643 """EventBus sink that records FSM loop events into the session database.
645 A single connection is opened at construction with ``check_same_thread``
646 disabled, since :meth:`send` may be called from the FSM thread while other
647 transports run their own threads; a lock serialises writes. Every
648 operation is best-effort — a database error is logged and swallowed so a
649 failing sink never aborts a loop run (the four ``wire_transports`` call
650 sites depend on this).
651 """
653 def __init__(self, db_path: Path | str = DEFAULT_DB_PATH) -> None:
654 self._path = Path(db_path)
655 self._lock = threading.Lock()
656 self._conn: sqlite3.Connection | None = None
657 try:
658 ensure_db(self._path)
659 self._conn = sqlite3.connect(str(self._path), check_same_thread=False)
660 except sqlite3.Error:
661 logger.warning(
662 "SQLiteTransport: could not open %s; sink disabled", self._path, exc_info=True
663 )
664 self._conn = None
666 def send(self, event: dict[str, Any]) -> None:
667 """Record a recognised event as a ``loop_events`` or ``issue_events`` row (best-effort)."""
668 conn = self._conn
669 if conn is None:
670 return
671 event_type = str(event.get("event", ""))
672 ts = str(event.get("ts") or _now())
673 try:
674 with self._lock:
675 if event_type in _LOOP_EVENT_TYPES:
676 loop_name = str(event.get("loop_name", "")) or None
677 state = event.get("state")
678 if event_type == "loop_complete":
679 state = event.get("outcome", state)
680 retries = event.get("retries")
681 conn.execute(
682 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
683 "VALUES(?, ?, ?, ?, ?)",
684 (
685 ts,
686 loop_name,
687 str(state) if state is not None else None,
688 event_type,
689 int(retries) if isinstance(retries, int) else None,
690 ),
691 )
692 _index(
693 conn,
694 content=" ".join(
695 str(p) for p in (loop_name, state, event_type) if p is not None
696 ),
697 kind="loop",
698 ref=loop_name or "",
699 anchor=f".loops/{loop_name}.yaml" if loop_name else "",
700 ts=ts,
701 )
702 elif event_type.startswith("issue."):
703 issue_id = event.get("issue_id")
704 transition = _derive_transition(event_type)
705 conn.execute(
706 "INSERT OR IGNORE INTO issue_events("
707 "ts, issue_id, transition, discovered_by, "
708 "issue_type, priority, captured_at, completed_at"
709 ") VALUES(?,?,?,?,?,?,?,?)",
710 (
711 ts,
712 issue_id,
713 transition,
714 event.get("discovered_by"),
715 event.get("issue_type"),
716 event.get("priority"),
717 event.get("captured_at"),
718 event.get("completed_at"),
719 ),
720 )
721 _index(
722 conn,
723 content=f"{issue_id or ''} {event.get('issue_type', '')}".strip(),
724 kind="issue",
725 ref=str(issue_id or ""),
726 anchor=event.get("issue_file", ""),
727 ts=ts,
728 )
729 else:
730 return
731 conn.commit()
732 except sqlite3.Error:
733 logger.warning("SQLiteTransport: write failed for event %r", event_type, exc_info=True)
735 def close(self) -> None:
736 """Close the underlying connection (best-effort)."""
737 if self._conn is not None:
738 try:
739 self._conn.close()
740 except sqlite3.Error:
741 pass
742 self._conn = None
745# ---------------------------------------------------------------------------
746# Backfill
747# ---------------------------------------------------------------------------
750def _hash_args(value: Any) -> str:
751 """Return a short stable hash of a tool-call argument structure."""
752 try:
753 blob = json.dumps(value, sort_keys=True, default=str)
754 except (TypeError, ValueError):
755 blob = repr(value)
756 return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
759_FILENAME_TYPE_RE = re.compile(r"(BUG|ENH|FEAT|EPIC)-(\d+)")
760_FILENAME_PRIORITY_RE = re.compile(r"^(P\d)")
763def _derive_type_priority(filename: str, fm: dict[str, Any]) -> tuple[str | None, str | None]:
764 """Derive (issue_type, priority) preferring frontmatter, falling back to filename.
766 Mirrors :func:`little_loops.issue_history.parsing.parse_completed_issue`'s
767 filename-parsing convention (``P[0-5]-[TYPE]-[NNN]-...``).
768 """
769 fm_type = fm.get("type")
770 issue_type: str | None = str(fm_type) if isinstance(fm_type, str) and fm_type else None
771 fm_priority = fm.get("priority")
772 priority: str | None = (
773 str(fm_priority) if isinstance(fm_priority, str) and fm_priority else None
774 )
775 if issue_type is None:
776 m = _FILENAME_TYPE_RE.search(filename)
777 if m:
778 issue_type = m.group(1)
779 if priority is None:
780 m = _FILENAME_PRIORITY_RE.match(filename)
781 if m:
782 priority = m.group(1)
783 return issue_type, priority
786def _backfill_issues(conn: sqlite3.Connection, issues_dir: Path) -> int:
787 """Seed ``issue_events`` from issue-file frontmatter under *issues_dir*.
789 Populates the v2 summary columns (``issue_type``, ``priority``,
790 ``completed_date``, ``captured_at``, ``completed_at``) so ``ll-history
791 summary`` can be answered from the DB without re-reading the files
792 (ENH-1621). ``completed_date`` is derived from ``completed_at`` (taking the
793 date portion) when present, leaving file-mtime / Resolution-section
794 inference to the file-parsing fallback path.
795 """
796 from little_loops.frontmatter import parse_frontmatter
798 count = 0
799 for issue_file in sorted(issues_dir.rglob("*.md")):
800 try:
801 fm = parse_frontmatter(issue_file.read_text(encoding="utf-8"))
802 except OSError:
803 continue
804 issue_id = fm.get("id")
805 if not issue_id:
806 m = _FILENAME_TYPE_RE.search(issue_file.name)
807 if m:
808 issue_id = f"{m.group(1)}-{m.group(2)}"
809 if not issue_id:
810 continue
811 status = str(fm.get("status", "open"))
812 discovered_by = fm.get("discovered_by")
813 captured_at = fm.get("captured_at")
814 completed_at = fm.get("completed_at")
815 ts = str(completed_at or captured_at or fm.get("discovered_date") or "")
816 issue_type, priority = _derive_type_priority(issue_file.name, fm)
817 completed_date: str | None = None
818 if isinstance(completed_at, str) and completed_at:
819 completed_date = completed_at[:10]
820 conn.execute(
821 "INSERT OR IGNORE INTO issue_events("
822 "ts, issue_id, transition, discovered_by, "
823 "issue_type, priority, completed_date, captured_at, completed_at"
824 ") VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
825 (
826 ts,
827 str(issue_id),
828 status,
829 str(discovered_by) if discovered_by else None,
830 issue_type,
831 priority,
832 completed_date,
833 str(captured_at) if captured_at else None,
834 str(completed_at) if completed_at else None,
835 ),
836 )
837 _index(
838 conn,
839 content=f"{issue_id} {status} {issue_type or ''}",
840 kind="issue",
841 ref=str(issue_id),
842 anchor=str(issue_file),
843 ts=ts,
844 )
845 count += 1
846 return count
849def _backfill_loops(conn: sqlite3.Connection, loops_dir: Path) -> int:
850 """Seed ``loop_events`` from FSM state JSON under ``.loops/.running`` + ``.history``."""
851 count = 0
852 for sub in (".running", ".history"):
853 directory = loops_dir / sub
854 if not directory.is_dir():
855 continue
856 for state_file in sorted(directory.glob("*.json")):
857 try:
858 data = json.loads(state_file.read_text(encoding="utf-8"))
859 except (OSError, json.JSONDecodeError):
860 continue
861 if not isinstance(data, dict):
862 continue
863 loop_name = str(data.get("loop_name") or state_file.stem)
864 state = data.get("current_state") or data.get("state")
865 ts = str(data.get("updated_at") or data.get("started_at") or "")
866 conn.execute(
867 "INSERT INTO loop_events(ts, loop_name, state, transition, retries) "
868 "VALUES(?, ?, ?, ?, ?)",
869 (ts, loop_name, str(state) if state else None, "backfill", None),
870 )
871 _index(
872 conn,
873 content=f"{loop_name} {state or ''}",
874 kind="loop",
875 ref=loop_name,
876 anchor=str(state_file),
877 ts=ts,
878 )
879 count += 1
880 return count
883def _backfill_tool_events(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
884 """Seed ``tool_events`` from assistant tool-use blocks in session JSONL files."""
885 count = 0
886 for jsonl_file in jsonl_files:
887 try:
888 handle = jsonl_file.open(encoding="utf-8")
889 except OSError:
890 continue
891 with handle:
892 for line in handle:
893 line = line.strip()
894 if not line:
895 continue
896 try:
897 record = json.loads(line)
898 except json.JSONDecodeError:
899 continue
900 if record.get("type") != "assistant":
901 continue
902 session_id = record.get("sessionId")
903 ts = str(record.get("timestamp") or "")
904 content = record.get("message", {}).get("content", [])
905 if not isinstance(content, list):
906 continue
907 for block in content:
908 if not isinstance(block, dict) or block.get("type") != "tool_use":
909 continue
910 tool_name = str(block.get("name", ""))
911 args = block.get("input", {})
912 conn.execute(
913 "INSERT INTO tool_events(ts, session_id, tool_name, args_hash, "
914 "result_size, bytes_in, bytes_out, cache_hit) "
915 "VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
916 (ts, session_id, tool_name, _hash_args(args), None, None, None, None),
917 )
918 _index(
919 conn,
920 content=tool_name,
921 kind="tool",
922 ref=tool_name,
923 anchor=str(jsonl_file),
924 ts=ts,
925 )
926 count += 1
927 return count
930def _backfill_messages(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
931 """Seed ``message_events`` from user blocks in session JSONL files.
933 Mirrors :func:`_backfill_tool_events` but selects ``type == "user"`` records
934 and inserts the user's textual content. Used by analyze_workflows() so
935 workflow analysis can read message bodies from the DB instead of a JSONL
936 file (ENH-1621).
937 """
938 count = 0
939 for jsonl_file in jsonl_files:
940 try:
941 handle = jsonl_file.open(encoding="utf-8")
942 except OSError:
943 continue
944 with handle:
945 for line in handle:
946 line = line.strip()
947 if not line:
948 continue
949 try:
950 record = json.loads(line)
951 except json.JSONDecodeError:
952 continue
953 if record.get("type") != "user":
954 continue
955 session_id = record.get("sessionId")
956 ts = str(record.get("timestamp") or "")
957 # The user message body lives at message.content; it may be a
958 # plain string or a list of content blocks. We persist a text
959 # rendering — list blocks are concatenated by their "text"
960 # field so analyze_workflows() can run its regexes over it.
961 content = record.get("message", {}).get("content", "")
962 if isinstance(content, list):
963 parts: list[str] = []
964 for block in content:
965 if isinstance(block, dict) and isinstance(block.get("text"), str):
966 parts.append(block["text"])
967 text = "\n".join(parts)
968 elif isinstance(content, str):
969 text = content
970 else:
971 text = ""
972 if not text.strip():
973 continue
974 conn.execute(
975 "INSERT INTO message_events(ts, session_id, content) VALUES(?, ?, ?)",
976 (ts, str(session_id) if session_id else None, text),
977 )
978 _index(
979 conn,
980 content=text[:512],
981 kind="message",
982 ref=str(session_id) if session_id else "",
983 anchor=str(jsonl_file),
984 ts=ts,
985 )
986 count += 1
987 return count
990def mine_corrections_from_messages(conn: sqlite3.Connection, config: dict | None = None) -> int:
991 """Scan ``message_events`` and insert matching rows into ``user_corrections``.
993 Designed for both the one-time retroactive pass over existing rows and
994 repeated calls during backfill; idempotent via ``INSERT OR IGNORE`` +
995 ``idx_corrections_dedup``. Only writes a ``search_index`` entry when the
996 row is actually inserted (rowcount == 1) to avoid duplicate FTS rows.
997 Gated by ``analytics.capture.corrections`` (ENH-1841).
999 Returns the count of newly inserted correction rows.
1000 """
1001 extra_patterns: list[str] = []
1002 if config is not None:
1003 from little_loops.config.features import AnalyticsCaptureConfig
1005 capture = AnalyticsCaptureConfig.from_dict(config.get("analytics", {}).get("capture", {}))
1006 if not capture.corrections:
1007 return 0
1008 extra_patterns = capture.correction_patterns
1010 count = 0
1011 rows = conn.execute("SELECT ts, session_id, content FROM message_events").fetchall()
1012 for ts, session_id, content in rows:
1013 if not content or not is_correction(content, extra_patterns=extra_patterns):
1014 continue
1015 text = content[:512]
1016 cursor = conn.execute(
1017 "INSERT OR IGNORE INTO user_corrections(ts, session_id, content, source)"
1018 " VALUES(?, ?, ?, 'backfill')",
1019 (ts, session_id, text),
1020 )
1021 if cursor.rowcount:
1022 _index(
1023 conn,
1024 content=text,
1025 kind="correction",
1026 ref=session_id or "",
1027 anchor="backfill",
1028 ts=ts,
1029 )
1030 count += 1
1031 return count
1034# ---------------------------------------------------------------------------
1035# Compaction — LCM-style summary DAG (FEAT-1712)
1036# ---------------------------------------------------------------------------
1039def _estimate_tokens(text: str) -> int:
1040 """Rough token estimate using the LCM convention: 4 characters per token."""
1041 return len(text) // 4
1044def _summarize_block(
1045 messages: list[str],
1046 budget: int,
1047 *,
1048 model: str | None = None,
1049 timeout: int = 60,
1050) -> str:
1051 """Summarize block_text to fit within budget tokens, with convergence guarantee.
1053 LCM Algorithm 3 three-level escalation:
1055 1. **Level 1**: Normal LLM summary (preserve details), target = budget.
1056 Accepted only if ``_estimate_tokens(result) < _estimate_tokens(input)``.
1057 2. **Level 2**: Aggressive bullet-point LLM summary at ``budget // 2``.
1058 Triggered when level-1 output is not smaller than input.
1059 3. **Level 3**: Deterministic truncation — ``min(budget * 4, 2048)`` characters.
1060 Guaranteed to produce output ≤ input by construction.
1061 Escalations are logged at WARNING level for operator visibility.
1062 """
1064 block_text = "\n---\n".join(messages)
1066 est_input = _estimate_tokens(block_text)
1068 # Short-circuit: for very small inputs an LLM summary cannot be meaningfully
1069 # smaller than the input — skip directly to deterministic truncation.
1070 if est_input < 25:
1071 return block_text[: min(budget * 4, 2048)]
1073 # -- Level 1: normal prose summary -------------------------------------------------
1074 level1_prompt = (
1075 "Summarize these session messages concisely (2-3 paragraphs), capturing key "
1076 "topics, decisions, and outcomes. Target approximately "
1077 f"{budget} tokens:\n\n" + block_text
1078 )
1079 result = _call_llm_for_summary(level1_prompt, model=model, timeout=timeout)
1080 if result and _estimate_tokens(result) < est_input:
1081 return result
1083 # -- Level 2: aggressive bullet-point summary at half budget -----------------------
1084 if result:
1085 logger.warning(
1086 "_summarize_block: level-1 summary not smaller than input "
1087 "(est_output=%d >= est_input=%d); escalating to level 2",
1088 _estimate_tokens(result),
1089 est_input,
1090 )
1091 else:
1092 logger.warning("_summarize_block: level-1 LLM call failed; escalating to level 2")
1093 level2_budget = max(budget // 2, 64)
1094 level2_prompt = (
1095 "Summarize these session messages as a compact bullet list. Be extremely terse: "
1096 "one line per key point, no preamble or commentary. Target approximately "
1097 f"{level2_budget} tokens:\n\n" + block_text
1098 )
1099 result = _call_llm_for_summary(level2_prompt, model=model, timeout=timeout)
1100 if result and _estimate_tokens(result) < est_input:
1101 return result
1103 # -- Level 3: deterministic truncation (guaranteed convergence) --------------------
1104 if result:
1105 logger.warning(
1106 "_summarize_block: level-2 summary not smaller than input "
1107 "(est_output=%d >= est_input=%d); escalating to level 3",
1108 _estimate_tokens(result),
1109 est_input,
1110 )
1111 else:
1112 logger.warning("_summarize_block: level-2 LLM call failed; escalating to level 3")
1113 # Truncation: min(budget * 4, 2048) chars. The 2048 cap (~512 tokens at 4 chars/token)
1114 # follows the LCM paper's level-3 constant, providing a strict convergence guarantee.
1115 max_chars = min(budget * 4, 2048)
1116 return block_text[:max_chars]
1119def _call_llm_for_summary(
1120 prompt: str,
1121 *,
1122 model: str | None = None,
1123 timeout: int = 60,
1124) -> str | None:
1125 """Call the host LLM for a summary and extract the prose ``result`` field.
1127 Returns the extracted prose string on success, or ``None`` if the LLM call
1128 failed or produced an unparseable response (allowing escalation logic to
1129 fall through to the next level).
1130 """
1132 try:
1133 inv = resolve_host().build_blocking_json(prompt=prompt, model=model)
1134 proc = subprocess.run(
1135 [inv.binary, *inv.args],
1136 capture_output=True,
1137 text=True,
1138 timeout=timeout,
1139 )
1140 except subprocess.TimeoutExpired:
1141 logger.warning("_call_llm_for_summary: LLM call timed out after %ds", timeout)
1142 return None
1143 except FileNotFoundError:
1144 logger.error(
1145 "_call_llm_for_summary: %s CLI not found. Install the active host CLI "
1146 "(see LL_HOST_CLI).",
1147 inv.binary,
1148 )
1149 return None
1151 if proc.returncode != 0:
1152 stderr_preview = proc.stderr.strip()[:200] if proc.stderr else "(no stderr)"
1153 logger.error(
1154 "_call_llm_for_summary: %s CLI returned exit code %d (stderr: %s)",
1155 inv.binary,
1156 proc.returncode,
1157 stderr_preview,
1158 )
1159 return None
1161 if not proc.stdout.strip():
1162 stderr_info = proc.stderr.strip()[:200] if proc.stderr else ""
1163 logger.error(
1164 "_call_llm_for_summary: %s CLI returned empty stdout on exit 0"
1165 + (f" (stderr: {stderr_info})" if stderr_info else "")
1166 )
1167 return None
1169 # Parse the JSON envelope and extract the 'result' field — see
1170 # evaluate_llm_structured() at fsm/evaluators.py:832-880 for the
1171 # canonical envelope-parsing pattern.
1172 try:
1173 stdout = proc.stdout.strip()
1174 try:
1175 envelope = json.loads(stdout)
1176 except json.JSONDecodeError:
1177 # Try JSONL: take the last non-empty line
1178 lines = [line for line in stdout.split("\n") if line.strip()]
1179 if not lines:
1180 raise
1181 envelope = json.loads(lines[-1])
1183 # Check for structured-output retry exhaustion or legacy is_error
1184 if envelope.get("subtype") == "error_max_structured_output_retries":
1185 logger.error(
1186 "_call_llm_for_summary: %s CLI could not produce valid output after retries",
1187 inv.binary,
1188 )
1189 return None
1190 if envelope.get("is_error", False):
1191 err_text = str(envelope.get("result", "") or "")[:200]
1192 logger.error(
1193 "_call_llm_for_summary: %s CLI reported error: %s",
1194 inv.binary,
1195 err_text,
1196 )
1197 return None
1199 # Extract the result field (plain prose; no --json-schema here)
1200 result = envelope.get("result", "")
1201 if not result:
1202 logger.error(
1203 "_call_llm_for_summary: empty result field in %s CLI response",
1204 inv.binary,
1205 )
1206 return None
1207 return str(result)
1209 except (json.JSONDecodeError, TypeError, ValueError) as e:
1210 raw_preview = proc.stdout[:300] if proc.stdout else "(empty)"
1211 logger.error(
1212 "_call_llm_for_summary: failed to parse LLM response: %s (raw: %s)",
1213 e,
1214 raw_preview,
1215 )
1216 return None
1219def _compact_session_conn(
1220 conn: sqlite3.Connection,
1221 session_id: str,
1222 budget: int = 4096,
1223 *,
1224 model: str | None = None,
1225 timeout: int = 60,
1226) -> int:
1227 """Compact one session using an existing connection. Returns new leaf node count.
1229 Greedy single-pass block grouping: token estimate ``len(s) // 4``. Each block
1230 gets one ``leaf`` summary_node; if the session accumulates ≥ 2 leaves a single
1231 ``condensed`` node is inserted (or silently skipped if one already exists via
1232 ``INSERT OR IGNORE`` + ``idx_summary_nodes_condensed_dedup``). Leaf dedup is
1233 handled by ``idx_summary_nodes_leaf_dedup`` on ``(session_id, ts_start, ts_end)``.
1234 """
1235 rows = conn.execute(
1236 "SELECT id, ts, content FROM message_events WHERE session_id = ? ORDER BY ts, id",
1237 (session_id,),
1238 ).fetchall()
1240 if not rows:
1241 return 0
1243 # Greedy block accumulation
1244 blocks: list[list[tuple[int, str, str]]] = []
1245 current: list[tuple[int, str, str]] = []
1246 current_tokens = 0
1248 for row in rows:
1249 msg_id, ts, content = row[0], row[1], row[2] or ""
1250 tok = _estimate_tokens(content)
1251 if current_tokens + tok > budget and current:
1252 blocks.append(current)
1253 current = [(msg_id, ts, content)]
1254 current_tokens = tok
1255 else:
1256 current.append((msg_id, ts, content))
1257 current_tokens += tok
1258 if current:
1259 blocks.append(current)
1261 now = _now()
1262 new_leaves = 0
1264 for block in blocks:
1265 ts_start = block[0][1]
1266 ts_end = block[-1][1]
1267 msg_ids = [r[0] for r in block]
1268 contents = [r[2] for r in block]
1270 summary = _summarize_block(contents, budget, model=model, timeout=timeout)
1271 cursor = conn.execute(
1272 "INSERT OR IGNORE INTO summary_nodes"
1273 "(kind, content, tokens, session_id, ts_start, ts_end, created_at)"
1274 " VALUES('leaf', ?, ?, ?, ?, ?, ?)",
1275 (summary, _estimate_tokens(summary), session_id, ts_start, ts_end, now),
1276 )
1277 if cursor.rowcount:
1278 leaf_id = cursor.lastrowid
1279 conn.executemany(
1280 "INSERT OR IGNORE INTO summary_spans(summary_id, message_event_id) VALUES(?, ?)",
1281 [(leaf_id, mid) for mid in msg_ids],
1282 )
1283 new_leaves += 1
1285 # Condensed node: one per session, summarises all leaves.
1286 all_leaves = conn.execute(
1287 "SELECT id, content FROM summary_nodes"
1288 " WHERE kind='leaf' AND session_id=? ORDER BY ts_start",
1289 (session_id,),
1290 ).fetchall()
1292 if len(all_leaves) >= 2:
1293 leaf_summaries = [r[1] for r in all_leaves]
1294 condensed_text = _summarize_block(leaf_summaries, budget, model=model, timeout=timeout)
1295 cursor = conn.execute(
1296 "INSERT OR IGNORE INTO summary_nodes"
1297 "(kind, content, tokens, session_id, ts_start, ts_end, created_at)"
1298 " VALUES('condensed', ?, ?, ?, NULL, NULL, ?)",
1299 (condensed_text, _estimate_tokens(condensed_text), session_id, now),
1300 )
1301 if cursor.rowcount:
1302 condensed_id = cursor.lastrowid
1303 conn.execute(
1304 "UPDATE summary_nodes SET parent_id = ?"
1305 " WHERE session_id = ? AND kind = 'leaf' AND parent_id IS NULL",
1306 (condensed_id, session_id),
1307 )
1309 return new_leaves
1312def _compact_sessions(
1313 conn: sqlite3.Connection,
1314 config: dict | None = None,
1315) -> int:
1316 """Compact all sessions in the sessions table; returns total new leaf nodes created.
1318 Gated by ``history.compaction.enabled`` (default ``false``). Skips silently when
1319 disabled so backfill() callers that omit config are unaffected.
1320 """
1321 from little_loops.config.features import CompactionConfig
1323 raw = config.get("history", {}).get("compaction", {}) if config else {}
1324 compact_cfg = CompactionConfig.from_dict(raw)
1325 if not compact_cfg.enabled:
1326 return 0
1328 rows = conn.execute("SELECT session_id FROM sessions").fetchall()
1329 total = 0
1330 for row in rows:
1331 total += _compact_session_conn(
1332 conn,
1333 row[0],
1334 budget=compact_cfg.budget_tokens,
1335 model=compact_cfg.model,
1336 timeout=compact_cfg.timeout,
1337 )
1338 return total
1341def compact_session(
1342 session_id: str,
1343 db: Path | str = DEFAULT_DB_PATH,
1344 *,
1345 config: dict | None = None,
1346) -> int:
1347 """Summarize message_events for one session into summary_nodes and summary_spans.
1349 Idempotent: repeated calls do not create duplicate nodes (INSERT OR IGNORE +
1350 partial unique indexes). Uses LCM Algorithm 3 three-level escalation (level 1:
1351 normal LLM summary → level 2: aggressive bullet-point LLM summary → level 3:
1352 deterministic truncation) so a leaf node is always produced. Returns the count
1353 of new leaf nodes created.
1354 """
1355 from little_loops.config.features import CompactionConfig
1357 raw = config.get("history", {}).get("compaction", {}) if config else {}
1358 compact_cfg = CompactionConfig.from_dict(raw)
1359 conn = connect(db)
1360 try:
1361 result = _compact_session_conn(
1362 conn,
1363 session_id,
1364 budget=compact_cfg.budget_tokens,
1365 model=compact_cfg.model,
1366 timeout=compact_cfg.timeout,
1367 )
1368 conn.commit()
1369 finally:
1370 conn.close()
1371 return result
1374def _backfill_sessions(conn: sqlite3.Connection, jsonl_files: list[Path]) -> int:
1375 """Seed ``sessions`` table by mapping each JSONL file to its session_id.
1377 Reads just enough of each file to extract the first ``sessionId`` value,
1378 then inserts one row per unique session. ``INSERT OR IGNORE`` + PRIMARY KEY
1379 makes repeated calls idempotent (ENH-1710).
1380 """
1381 count = 0
1382 for jsonl_file in jsonl_files:
1383 try:
1384 handle = jsonl_file.open(encoding="utf-8")
1385 except OSError:
1386 continue
1387 with handle:
1388 for line in handle:
1389 line = line.strip()
1390 if not line:
1391 continue
1392 try:
1393 record = json.loads(line)
1394 except json.JSONDecodeError:
1395 continue
1396 session_id = record.get("sessionId")
1397 if session_id:
1398 cur = conn.execute(
1399 "INSERT OR IGNORE INTO sessions(session_id, jsonl_path) VALUES(?, ?)",
1400 (str(session_id), str(jsonl_file)),
1401 )
1402 count += cur.rowcount
1403 break # one session_id per file is sufficient
1404 return count
1407def _mtime(path: Path) -> float:
1408 """Return file modification time as a Unix float, or 0.0 if inaccessible."""
1409 try:
1410 return path.stat().st_mtime
1411 except OSError:
1412 return 0.0
1415def backfill(
1416 db: Path | str = DEFAULT_DB_PATH,
1417 *,
1418 issues_dir: Path | None = None,
1419 loops_dir: Path | None = None,
1420 jsonl_files: list[Path] | None = None,
1421 config: dict | None = None,
1422) -> dict[str, int]:
1423 """Populate the database from existing on-disk sources.
1425 Reads issue-file frontmatter, FSM loop-state JSON, and (optionally) session
1426 JSONL tool-use blocks plus user-message blocks. Returns a per-kind count of
1427 rows inserted. Sources that are absent are skipped silently.
1428 """
1429 issues_dir = issues_dir if issues_dir is not None else Path(".issues")
1430 loops_dir = loops_dir if loops_dir is not None else Path(".loops")
1431 conn = connect(db)
1432 counts: dict[str, int] = {
1433 "issues": 0,
1434 "loops": 0,
1435 "tools": 0,
1436 "messages": 0,
1437 "sessions": 0,
1438 "corrections": 0,
1439 "summaries": 0,
1440 }
1441 try:
1442 if issues_dir.is_dir():
1443 counts["issues"] = _backfill_issues(conn, issues_dir)
1444 if loops_dir.is_dir():
1445 counts["loops"] = _backfill_loops(conn, loops_dir)
1446 if jsonl_files:
1447 counts["tools"] = _backfill_tool_events(conn, jsonl_files)
1448 counts["messages"] = _backfill_messages(conn, jsonl_files)
1449 counts["sessions"] = _backfill_sessions(conn, jsonl_files)
1450 counts["corrections"] = mine_corrections_from_messages(conn, config)
1451 counts["summaries"] = _compact_sessions(conn, config)
1452 conn.commit()
1453 finally:
1454 conn.close()
1455 return counts
1458def backfill_incremental(
1459 db: Path | str = DEFAULT_DB_PATH,
1460 *,
1461 jsonl_files: list[Path],
1462 since_ts: float | None = None,
1463 config: dict | None = None,
1464) -> dict[str, int]:
1465 """Backfill only JSONL files modified after *since_ts*.
1467 If *since_ts* is ``None``, reads ``last_backfill_ts`` from the ``meta``
1468 table (defaults to 0.0 — all files — when the key is absent or NULL).
1469 On success, writes the current UTC time as the new ``last_backfill_ts``
1470 so the next call automatically skips already-processed files.
1472 Issues and loop-state JSON are NOT backfilled here; this variant is
1473 JSONL-only and designed for low-latency background use in session hooks.
1474 Errors are not suppressed — the caller (session hook) catches them and
1475 logs a warning.
1476 """
1477 conn = connect(db)
1478 counts: dict[str, int] = {"tools": 0, "messages": 0, "sessions": 0, "corrections": 0}
1479 try:
1480 if since_ts is None:
1481 row = conn.execute("SELECT value FROM meta WHERE key = 'last_backfill_ts'").fetchone()
1482 raw = row[0] if (row and row[0]) else None
1483 if raw:
1484 try:
1485 since_ts = datetime.fromisoformat(str(raw).replace("Z", "+00:00")).timestamp()
1486 except ValueError:
1487 since_ts = 0.0
1488 else:
1489 since_ts = 0.0
1491 filtered = [f for f in jsonl_files if _mtime(f) >= since_ts]
1492 if filtered:
1493 counts["sessions"] = _backfill_sessions(conn, filtered)
1494 counts["tools"] = _backfill_tool_events(conn, filtered)
1495 counts["messages"] = _backfill_messages(conn, filtered)
1497 counts["corrections"] = mine_corrections_from_messages(conn, config)
1498 conn.execute(
1499 "INSERT INTO meta(key, value) VALUES('last_backfill_ts', ?) "
1500 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
1501 (_now(),),
1502 )
1503 conn.commit()
1504 finally:
1505 conn.close()
1506 return counts