JSONL을 기준 기록으로 두고 파생 저장소는 다시 만들어

기준 원본이 하나여야 복구 방향이 선명해

cwkPippa는 프롬프트와 델타, 도구 호출·결과, 최종 사용량을 conversation_id별 암호화 JSONL에 덧붙여. SQLite와 ChromaDB는 빠른 조회를 위한 파생 거울이며 JSONL에서 다시 만들 수 있어. 불일치가 생기면 여러 저장소를 손으로 맞추지 않고 파생물을 지운 뒤 원본을 재생해.

화면에 보이기 전에 디스크에 써

각 델타는 사용자 화면이나 API로 내보내기 전에 JSONL에 저장해야 해. 스트림이 중간에 죽어도 사용자가 본 모든 토큰이 기록에 남는 순서야. 줄 단위 쓰기와 명시적 flush를 함께 써 갑작스러운 SIGTERM에도 마지막 완성 줄을 잃지 않게 해.

읽을 때 불완전한 턴을 치유해

cwkPippa는 대화를 가져올 때 _heal_incomplete_turns를 실행해 델타에서 중단된 턴을 복원하고 시간 순서를 정리해. 기준 기록은 내구성이 있지만 파생 상태는 흔들릴 수 있으므로, 읽기 경로가 안전한 투영을 만들어 프런트엔드에 넘겨.

원칙: 기준 원본은 하나이고 내구성 있게 덧붙여야 해. 색인과 캐시는 수정할 원본이 아니라 다시 만들 파생물이야.

Code

Write before show·python

import json, pathlib

async def stream_with_persistence(conv_id: str, stream):
    p = pathlib.Path(f"/srv/sessions/{conv_id}.jsonl").open("a", buffering=1)
    try:
        async for event in stream:
            # Persist FIRST.
            p.write(json.dumps({"event": event.dict()}) + "\n")
            p.flush()
            # 그다음 user에 렌더.
            yield event
    finally:
        p.close()

Inconsistency에 JSONL에서 SQLite rebuild·python

import json

def rebuild_session(conn, conv_id: str, jsonl_path: str):
    # 이 세션의 derived 상태 퍼지.
    conn.execute("DELETE FROM messages WHERE conversation_id = ?", (conv_id,))
    # JSONL에서 replay.
    for line in open(jsonl_path):
        ev = json.loads(line)
        if ev["event"]["type"] == "message_complete":
            conn.execute(
                "INSERT INTO messages (conversation_id, role, content, created_at) VALUES (?, ?, ?, ?)",
                (conv_id, ev["event"]["role"], ev["event"]["text"], ev["event"]["created_at"]),
            )
    conn.commit()

기준 원본이 하나여야 복구 방향이 선명해

화면에 보이기 전에 디스크에 써

읽을 때 불완전한 턴을 치유해

Code

External links

Exercise

Progress

댓글 0