Session Management

~22 min · sessions, state, persistence

Level 0Tokenizer

0 XP0/54 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

A SessionStore provides file-backed persistence for conversations, enabling multi-session agents with token usage tracking.

JSONL ground truth, SQLite query cache

Every event (user message, assistant text delta, tool_call requested, tool_call result, errors) is written line-by-line to {conversation_id}.jsonl BEFORE the frontend sees it ('write before show'). SQLite holds a denormalized 'messages' table for fast querying. ChromaDB holds embeddings for semantic search.

The invariant: SQLite and ChromaDB are derived mirrors of JSONL. If they ever disagree, JSONL wins; rebuild the SQLite rows for that session by purge-and-replay. Never patch SQLite to match what you think happened — patch is where bugs live. Replay is reproducible.

Code

Session record schema (SQLite/JSONL)·python

import uuid, json
from pathlib import Path
from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class Session:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    conversation_history: list[dict] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    token_usage: dict = field(default_factory=lambda: {
        "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0
    })

class SessionStore:
    """File-backed session persistence."""
    def __init__(self, store_dir: Path = Path("./sessions")):
        self.store_dir = store_dir
        self.store_dir.mkdir(parents=True, exist_ok=True)
        self._sessions = {}

    def create(self, system_prompt=None) -> Session:
        session = Session()
        if system_prompt:
            session.conversation_history.append(
                {"role": "system", "content": system_prompt}
            )
        self._sessions[session.id] = session
        return session

    def save(self, session: Session):
        path = self.store_dir / f"{session.id}.json"
        with open(path, "w") as f:
            json.dump({"id": session.id,
                "conversation_history": session.conversation_history,
                "token_usage": session.token_usage}, f, indent=2)

    def update_usage(self, session, usage):
        for key in ("prompt_tokens", "completion_tokens", "total_tokens"):
            session.token_usage[key] += usage.get(key, 0)

External links

Append-only logs (Kafka philosophy)

Exercise

Build a session store that writes JSONL line-by-line and a SQLite index alongside. Add a 'rebuild_sqlite_from_jsonl(session_id)' that wipes the rows and replays. Verify it converges.