Session Management — JSONL ground truth

매 event (user message, assistant text delta, tool_call requested, tool_call result, error) 를 {conversation_id}.jsonl 에 line-by-line, frontend 보기 전 ('write before show'). SQLite 는 fast query 용 denormalized 'messages' table. ChromaDB 는 semantic search 용 embedding.

Invariant — JSONL 이 ground truth

SQLite 와 ChromaDB 는 derived mirror. JSONL 과 disagree 시 JSONL 이김. SQLite row 를 patch 하지 마 — purge-and-replay 로 rebuild. Patch 는 버그가 살아.

Append-only 의 의미

Line 추가만, 절대 modify/delete X. 사용자가 'undo' 해도 — 새 'undo event' append, original event 그대로. 모든 사고가 forensically 추적 가능, recovery 가 reproducible.

cwkPippa 의 session 모델

cwkPippa 는 JSONL line-by-line, frontend 보기 전 ('write before show'). SQLite 는 queryable row, ChromaDB 는 embedding. JSONL 이 authoritative — 두 downstream store 모두 replay 로 rebuild.

Encryption at rest

cwkPippa 는 JSONL 에 line-level Fernet encryption — keychain 에 저장된 passphrase keyed. Peer Mac 들은 opaque blob 으로 fleet rsync round-trip, plaintext 는 절대 X.

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)

Session Management — JSONL ground truth

Invariant — JSONL 이 ground truth

Append-only 의 의미

cwkPippa 의 session 모델

Encryption at rest

Code

External links

Exercise

Progress

댓글 0