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