cwkPippa learned this in production: every event (prompt, delta, tool use, tool result, final usage) lands in JSONL keyed by conversation_id, encrypted at rest, append-only. SQLite and ChromaDB are derived mirrors — fast indexes for the application, rebuildable from the JSONL. Recovery from any inconsistency is purge-and-replay, not patch-and-reconcile.
Write before show
The reliability invariant: persist each delta to JSONL before rendering it to the user or surfacing it via API. A crashed stream still has every visible token on disk. Pair line-buffered writes with explicit flush so a SIGTERM mid-write does not lose the last line.
Healing on read
cwkPippa runs _heal_incomplete_turns on every conversation GET — rebuilds aborted turns from delta events, normalizes timestamps so order is preserved. The frontend never sees a broken turn. Healing is the cost of letting JSONL be canonical: derived state can drift, and the read path must repair it.
Principle: Single source of truth, durable, append-only. Indexes and caches are derived. Recovery is rebuild, not patch.
Code
Write before show·python
import dataclasses, json, pathlib, time
def to_record(event) -> dict:
# Agent SDK messages are dataclasses; Anthropic SDK stream events are pydantic models.
if dataclasses.is_dataclass(event):
return {"type": type(event).__name__, **dataclasses.asdict(event)}
return event.model_dump(mode="json")
def append(conv_id: str, event: dict):
# One envelope per line, stamped at write time, flushed before anything else happens.
with pathlib.Path(f"/srv/sessions/{conv_id}.jsonl").open("a", buffering=1) as p:
p.write(json.dumps({"ts": time.time(), "event": event}, default=str) + "\n")
p.flush()
async def stream_with_persistence(conv_id: str, prompt: str, stream):
append(conv_id, {"type": "UserPrompt", "text": prompt}) # the prompt is part of the record too
async for event in stream:
append(conv_id, to_record(event)) # persist FIRST
yield event # then render to the user
Rebuild SQLite from JSONL on inconsistency·python
import json
def rebuild_session(conn, conv_id: str, jsonl_path: str):
# Purge derived state for this session.
conn.execute("DELETE FROM messages WHERE conversation_id = ?", (conv_id,))
# Replay from JSONL, reading exactly the envelope the writer wrote.
for line in open(jsonl_path):
rec = json.loads(line)
ev = rec["event"]
if ev["type"] == "UserPrompt":
role, text = "user", ev["text"]
elif ev["type"] == "AssistantMessage":
role = "assistant"
text = "".join(block["text"] for block in ev["content"] if "text" in block)
else:
continue # deltas, tool calls and results stay in the record, not in this table
conn.execute(
"INSERT INTO messages (conversation_id, role, content, created_at) VALUES (?, ?, ?, ?)",
(conv_id, role, text, rec["ts"]),
)
conn.commit()
For one chat-style feature, route every event to JSONL before rendering. Add a tool that rebuilds your DB rows for a session from the JSONL. Run rebuild against a known session and confirm the result matches the live DB.
Hint
If the rebuild differs from live, JSONL has bugs (or a write was missed). Fix the writer first; the rebuild is the integrity check.
Progress
Progress is local-only — sign in to sync across devices.