Coverage for src/lilbee/sessions/store.py: 100%
199 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Append-only JSONL store for chat sessions.
3Each session is one ``<id>.jsonl`` file under ``cfg.data_dir/sessions``. The file
4is a strictly append-only event log: one JSON object per line, appended and
5fsynced, never rewritten. Event types are ``meta`` (first line), ``title``
6(newest wins, so rename appends rather than rewrites), ``message``, and
7``summary`` (newest wins; compaction's condensed view of the turns that no
8longer fit the prompt). The only corruption an append log can suffer is a torn
9final line, which the reader skips.
10"""
12from __future__ import annotations
14import json
15import os
16from collections.abc import Callable, Iterator
17from dataclasses import dataclass
18from datetime import UTC, datetime
19from enum import StrEnum
20from pathlib import Path
21from typing import Any
22from uuid import uuid4
24from filelock import FileLock
26from lilbee.core.config import cfg
28SESSIONS_DIRNAME = "sessions"
29SESSIONS_DISABLED_HINT = (
30 "Sessions are off. Turn them on with /set sessions_enabled true in the TUI, "
31 "settings_set via MCP, or sessions_enabled = true in config.toml."
32)
33AGENT_SESSIONS_DISABLED_HINT = (
34 "Agent sessions are off. Turn them on with settings_set mcp_sessions_enabled "
35 "true, /set mcp_sessions_enabled true in the TUI, or mcp_sessions_enabled = "
36 "true in config.toml."
37)
38# Bounds a wedged lock holder; a healthy append holds the lock for milliseconds.
39_APPEND_LOCK_TIMEOUT_S = 10
40UNTITLED_SESSION_TITLE = "Untitled chat"
41TITLE_MAX_LEN = 60
42TITLE_ELLIPSIS = "…"
45def sessions_enabled() -> bool:
46 """True when session persistence is on for the human surfaces (default on)."""
47 return cfg.sessions_enabled
50def agent_sessions_enabled() -> bool:
51 """True when agent (MCP) sessions are on (default off).
53 Independent of ``sessions_enabled``: the two govern separate domains, so an
54 agent's working state can be off while a human's conversations are saved.
55 """
56 return cfg.mcp_sessions_enabled
59class SessionEventType(StrEnum):
60 """The tag on every line of a session log."""
62 META = "meta"
63 TITLE = "title"
64 MESSAGE = "message"
65 SUMMARY = "summary"
66 ORIGIN = "origin"
69class SessionOrigin(StrEnum):
70 """The surface a session belongs to: whoever created it, or was last
71 explicitly transferred to. Appends from any other surface are refused, so
72 an agent cannot splice its turns into a conversation a human owns."""
74 TUI = "tui"
75 MCP = "mcp"
76 HTTP = "http"
77 CLI = "cli"
80# The surfaces a human drives directly. Their sessions are one conversation
81# space (start in Obsidian, continue in the TUI); agent sessions are working
82# state and stay out of it unless asked for.
83HUMAN_ORIGINS: frozenset[SessionOrigin] = frozenset(
84 {SessionOrigin.TUI, SessionOrigin.HTTP, SessionOrigin.CLI}
85)
88class MessageRole(StrEnum):
89 """Author of a chat message."""
91 USER = "user"
92 ASSISTANT = "assistant"
95class TitleSource(StrEnum):
96 """Where a session title came from."""
98 AUTO = "auto"
99 CUSTOM = "custom"
102@dataclass(frozen=True)
103class SessionMessage:
104 """One turn in a session. ``ts`` is stamped by the store on write."""
106 role: MessageRole
107 content: str
108 sources: tuple[str, ...] = ()
109 ts: str = ""
112@dataclass(frozen=True)
113class SessionMeta:
114 """Session metadata, reconstructed from the log without its message bodies."""
116 id: str
117 title: str
118 created_at: str
119 updated_at: str
120 model_ref: str
121 scope: str
122 message_count: int
123 origin: SessionOrigin = SessionOrigin.TUI
124 """Owning surface. Files written before ownership existed carry no origin;
125 the only writer then was the TUI, so that is the fallback."""
128@dataclass(frozen=True)
129class Session:
130 """A session's metadata plus its full transcript."""
132 meta: SessionMeta
133 messages: tuple[SessionMessage, ...]
134 # Rolling summary of the turns compaction has folded away, empty until the
135 # conversation first outgrows the prompt budget. It lives here rather than on
136 # the meta because only replaying a session needs it: listing does not, and
137 # carrying a paragraph per session would bloat the drawer's hot path and
138 # every HTTP/MCP list payload.
139 summary: str = ""
142class SessionNotFoundError(Exception):
143 """Raised when a session id has no backing file."""
145 def __init__(self, session_id: str) -> None:
146 super().__init__(f"No session with id {session_id!r}")
147 self.session_id = session_id
150def _may_append(surface: SessionOrigin, owner: SessionOrigin) -> bool:
151 """Whether *surface* may append to a session owned by *owner*.
153 The human surfaces are one conversation space (the same person in the TUI,
154 Obsidian, or the shell), so they append to each other's sessions freely.
155 Agent sessions are working state: only the agent surface appends to them,
156 and it appends to nothing else without an explicit claim.
157 """
158 if surface is owner:
159 return True
160 return surface in HUMAN_ORIGINS and owner in HUMAN_ORIGINS
163class SessionOwnershipError(Exception):
164 """Raised when a surface appends to a session another surface owns."""
166 def __init__(self, session_id: str, owner: SessionOrigin, surface: SessionOrigin) -> None:
167 super().__init__(
168 f"Session {session_id!r} belongs to the {owner.value} surface; "
169 f"claim it before appending from {surface.value}."
170 )
171 self.session_id = session_id
172 self.owner = owner
173 self.surface = surface
176def derive_title(text: str) -> str:
177 """Title a session from its first user message: first line, truncated."""
178 stripped = text.strip()
179 if not stripped:
180 return UNTITLED_SESSION_TITLE
181 first = stripped.splitlines()[0]
182 if len(first) > TITLE_MAX_LEN:
183 return first[:TITLE_MAX_LEN] + TITLE_ELLIPSIS
184 return first
187def _message_from_event(event: dict[str, Any], ts: str) -> SessionMessage:
188 """Reconstruct one message from its ``message`` event line."""
189 return SessionMessage(
190 role=MessageRole(event["role"]),
191 content=event["content"],
192 sources=tuple(event.get("sources", [])),
193 ts=ts,
194 )
197class SessionStore:
198 """Reads and appends session logs under ``cfg.data_dir/sessions``.
200 The directory is resolved late-bound from ``cfg`` on every call, so the store
201 follows a reconfigured data dir (and test isolation) without reconstruction.
202 ``clock`` is injectable for deterministic tests.
203 """
205 def __init__(self, clock: Callable[[], datetime] | None = None) -> None:
206 self._clock = clock or (lambda: datetime.now(UTC))
207 # path -> (size, mtime, meta) from the last fold of that file; see _meta_for.
208 self._meta_cache: dict[Path, tuple[int, float, SessionMeta]] = {}
210 @property
211 def _dir(self) -> Path:
212 return cfg.data_dir / SESSIONS_DIRNAME
214 def _path(self, session_id: str) -> Path:
215 return self._dir / f"{session_id}.jsonl"
217 def _now(self) -> str:
218 return self._clock().isoformat()
220 def _require(self, session_id: str) -> Path:
221 path = self._path(session_id)
222 if not path.exists():
223 raise SessionNotFoundError(session_id)
224 return path
226 @staticmethod
227 def _write_event(path: Path, event: dict[str, Any]) -> None:
228 # Per-session lock: two writers on one id (a second process, another
229 # surface) serialize instead of interleaving lines. Appends take
230 # milliseconds, so a blocked writer waits, never fails, under any
231 # realistic contention; the timeout only bounds a wedged holder.
232 with (
233 FileLock(str(path) + ".lock", timeout=_APPEND_LOCK_TIMEOUT_S),
234 path.open("a", encoding="utf-8") as fh,
235 ):
236 fh.write(json.dumps(event) + "\n")
237 fh.flush()
238 os.fsync(fh.fileno())
240 @staticmethod
241 def _iter_events(path: Path) -> Iterator[dict[str, Any]]:
242 # A torn multi-byte character decodes here, outside the try that skips
243 # torn lines; replacing makes it a JSON failure that try can catch.
244 with path.open(encoding="utf-8", errors="replace") as fh:
245 for raw in fh:
246 line = raw.strip()
247 if not line:
248 continue
249 try:
250 yield json.loads(line)
251 except json.JSONDecodeError:
252 continue # torn final line; skip it
254 def create(self, model_ref: str, scope: str, origin: SessionOrigin = SessionOrigin.TUI) -> str:
255 """Start a new session owned by *origin* and return its id."""
256 session_id = uuid4().hex
257 self._dir.mkdir(parents=True, exist_ok=True)
258 now = self._now()
259 self._write_event(
260 self._path(session_id),
261 {
262 "type": SessionEventType.META,
263 "id": session_id,
264 "created_at": now,
265 "model_ref": model_ref,
266 "scope": scope,
267 "origin": origin,
268 "ts": now,
269 },
270 )
271 return session_id
273 def add_message(
274 self, session_id: str, message: SessionMessage, *, surface: SessionOrigin | None = None
275 ) -> None:
276 """Append one message event to an existing session.
278 With *surface* given, the append is refused unless that surface owns the
279 session (see ``transfer``); without it the caller is a library embedder
280 that manages its own store and ownership does not apply.
281 """
282 path = self._require(session_id)
283 if surface is not None:
284 meta = self._meta_for(path)
285 if meta is not None and not _may_append(surface, meta.origin):
286 raise SessionOwnershipError(session_id, meta.origin, surface)
287 self._write_event(
288 path,
289 {
290 "type": SessionEventType.MESSAGE,
291 "role": message.role,
292 "content": message.content,
293 "sources": list(message.sources),
294 "ts": self._now(),
295 },
296 )
298 def transfer(self, session_id: str, origin: SessionOrigin) -> None:
299 """Append an origin event handing the session to *origin*; newest wins.
301 This is the explicit bridge between the human and agent domains: an
302 agent claims a session whose id the user handed it, and POST /claim
303 brings one back. Never implicit in an append.
304 """
305 self._write_event(
306 self._require(session_id),
307 {"type": SessionEventType.ORIGIN, "origin": origin, "ts": self._now()},
308 )
310 def set_title(self, session_id: str, title: str, source: TitleSource) -> None:
311 """Append a title event; the newest title wins on read."""
312 self._write_event(
313 self._require(session_id),
314 {"type": SessionEventType.TITLE, "title": title, "source": source, "ts": self._now()},
315 )
317 def set_summary(self, session_id: str, summary: str) -> None:
318 """Append a summary event; the newest summary wins on read.
320 Compaction folds the oldest turns into a summary once they no longer fit
321 the prompt. The messages themselves stay in the log untouched: the
322 transcript the user scrolls is always complete, and only what is fed to
323 the model is condensed.
324 """
325 self._write_event(
326 self._require(session_id),
327 {"type": SessionEventType.SUMMARY, "summary": summary, "ts": self._now()},
328 )
330 def delete(self, session_id: str) -> None:
331 """Remove a session's file."""
332 self._require(session_id).unlink()
334 def get(self, session_id: str) -> Session:
335 """Replay a session's log into its reconstructed view."""
336 meta, messages, summary = self._replay(
337 session_id, self._require(session_id), collect_messages=True
338 )
339 return Session(meta=meta, messages=messages, summary=summary)
341 def list(self, origins: frozenset[SessionOrigin] | None = None) -> list[SessionMeta]:
342 """Sessions' metadata, newest first; *origins* narrows to those surfaces.
344 Listing replays every event of every session, so it is the one hot path
345 here: the drawer runs it on open. Messages are not materialised (only
346 counted), and each file's meta is memoised against its size and mtime so
347 reopening a vault that has not changed costs one stat() per session.
348 """
349 if not self._dir.exists():
350 return []
351 paths = list(self._dir.glob("*.jsonl"))
352 metas = [meta for meta in (self._meta_for(path) for path in paths) if meta is not None]
353 if origins is not None:
354 metas = [meta for meta in metas if meta.origin in origins]
355 # Drop cache entries for sessions that no longer exist, so a long-lived
356 # store does not pin the meta of every session ever deleted.
357 live = {path for path in paths}
358 self._meta_cache = {p: v for p, v in self._meta_cache.items() if p in live}
359 return sorted(metas, key=lambda meta: (meta.updated_at, meta.id), reverse=True)
361 def _meta_for(self, path: Path) -> SessionMeta | None:
362 """Meta for one session file, reusing the last fold when it is unchanged.
364 The log is append-only, so any new event grows the file: size plus mtime
365 is enough to notice a change. A file that grows between the stat and the
366 read is simply re-folded on the next list(), never served stale.
368 Returns None when the file goes away underneath us, which is routine: the
369 CLI or another surface can delete a session while the drawer is listing.
370 Reading it instead would raise straight out of list().
371 """
372 try:
373 stat = path.stat()
374 except OSError:
375 return None
376 cached = self._meta_cache.get(path)
377 if cached is not None and cached[0] == stat.st_size and cached[1] == stat.st_mtime:
378 return cached[2]
379 try:
380 meta = self._replay(path.stem, path, collect_messages=False)[0]
381 except OSError:
382 return None
383 self._meta_cache[path] = (stat.st_size, stat.st_mtime, meta)
384 return meta
386 def _replay(
387 self, session_id: str, path: Path, *, collect_messages: bool
388 ) -> tuple[SessionMeta, tuple[SessionMessage, ...], str]:
389 """Fold a session's event log into its meta, messages and summary.
391 ``collect_messages=False`` is for listing, which needs only the count:
392 building a SessionMessage per message across a whole vault is pure waste.
393 """
394 created_at = ""
395 model_ref = ""
396 scope = ""
397 title = UNTITLED_SESSION_TITLE
398 updated_at = ""
399 summary = ""
400 origin = SessionOrigin.TUI
401 message_count = 0
402 messages: list[SessionMessage] = []
403 for event in self._iter_events(path):
404 ts = event.get("ts", "")
405 updated_at = ts
406 event_type = event.get("type")
407 if event_type == SessionEventType.META:
408 created_at = event["created_at"]
409 model_ref = event["model_ref"]
410 scope = event["scope"]
411 origin = SessionOrigin(event.get("origin", SessionOrigin.TUI))
412 elif event_type == SessionEventType.ORIGIN:
413 origin = SessionOrigin(event["origin"])
414 elif event_type == SessionEventType.TITLE:
415 title = event["title"]
416 elif event_type == SessionEventType.SUMMARY:
417 summary = event["summary"]
418 elif event_type == SessionEventType.MESSAGE:
419 message_count += 1
420 if collect_messages:
421 messages.append(_message_from_event(event, ts))
422 meta = SessionMeta(
423 id=session_id,
424 title=title,
425 created_at=created_at,
426 updated_at=updated_at,
427 model_ref=model_ref,
428 scope=scope,
429 message_count=message_count,
430 origin=origin,
431 )
432 return meta, tuple(messages), summary