본문 바로가기
C.W.K.
Stream
Lesson 03 of 07 · published

Session Management — JSONL ground truth

~22 min · sessions, state, persistence

Level 0Tokenizer
0 XP0/54 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

user message, assistant text delta, tool_call 요청과 결과, error를 {conversation_id}.jsonl에 한 줄씩 기록하고 나서 frontend에 보여줘. SQLite의 denormalized messages table은 빠른 query용이고, ChromaDB는 semantic search용 embedding이야.

JSONL을 원본으로 삼아

SQLite와 ChromaDB는 JSONL에서 만든 파생 mirror야. 내용이 다르면 JSONL을 기준으로 SQLite를 비우고 replay해 다시 만들어. 잘못된 row만 patch하면 원인이 남아 다음 복구 때 다시 드러나.

append-only는 과거를 고치지 않는다는 뜻이야

기존 line을 수정하거나 지우지 말고 새 event만 추가해. 사용자가 undo해도 원래 event는 남기고 undo event를 덧붙여. 그러면 사고 경로를 추적하고 같은 방식으로 복구할 수 있어.

cwkPippa의 session 모델

cwkPippa는 JSONL line을 frontend에 보여주기 전에 기록해. SQLite query row와 ChromaDB embedding은 모두 JSONL을 replay해 다시 만들 수 있어.

저장된 JSONL도 암호화해

cwkPippa는 keychain에 보관한 passphrase로 JSONL 각 line을 Fernet 암호화해. 다른 Mac에는 opaque blob 상태로 rsync되고 plaintext 파일로 남지 않아.

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

Exercise

JSONL line 기록과 SQLite index를 함께 쓰는 session store를 만들어. rebuild_sqlite_from_jsonl(session_id)를 추가해 row를 모두 지우고 replay한 뒤 같은 상태로 수렴하는지 확인해.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.