Durable Event Log

Persist lifecycle events to SQLite and query them per session, even after restart

Attach a durable sink to the event bus so every lifecycle event survives restart and reads back per session, in order.

from praisonaiagents import Agent
from praisonaiagents.bus import get_default_bus, SqliteEventLog

bus = get_default_bus()
bus.attach_sink(SqliteEventLog())            # opt-in; nothing persists without this

agent = Agent(name="Assistant", instructions="Be helpful")
agent.start("Hello!")

Nothing is persisted by default — a bare Agent() attaches no sink and the zero-overhead in-memory path stays untouched. You opt in by attaching a sink.

Quick Start

1. Attach a sink and publish

   from praisonaiagents.bus import EventBus, EventType, SqliteEventLog
   
   bus = EventBus()
   log = SqliteEventLog(db_path="./events.sqlite")
   bus.attach_sink(log)
   
   # Even with no subscribers, the sink receives the event.
   bus.publish(EventType.SESSION_CREATED, {"session_id": "s1"})
   bus.publish(EventType.MESSAGE_CREATED, {"session_id": "s1", "text": "hi"})
   
   for ev in log.query("s1"):
       print(ev.metadata["seq"], ev.type, ev.data)
   # 1 session_created {'session_id': 's1'}
   # 2 message_created {'session_id': 's1', 'text': 'hi'}

2. Read the timeline after a restart

   from praisonaiagents.bus import SqliteEventLog
   
   # Re-open the same file in a fresh process.
   log = SqliteEventLog(db_path="./events.sqlite")
   timeline = log.query("s1")
   log.close()

3. Walk the timeline with a cursor

   last_seq = 0
   while True:
       batch = log.query("s1", after_seq=last_seq, limit=100)
       if not batch:
           break
       for ev in batch:
           handle(ev)
       last_seq = batch[-1].metadata["seq"]

How It Works

The log persists the same Event / EventType shapes as the in-memory bus — one row per event, keyed by session_id with a monotonic per-session seq. Writes are best-effort (a broken sink never breaks the turn), on-disk databases use WAL, and prune keeps the file bounded.

Sinks resolve the session from event.data["session_id"], then event.metadata["session_id"], then event.source. With none present, the event is stored under an empty session id.

Configuration Options

API

Bus wiring:

Common Patterns

Timeline export:

for ev in log.query("s1", limit=10_000):
    print(ev.metadata["seq"], ev.type, ev.data)

Periodic prune:

log.prune(older_than_days=30, max_rows=1_000_000)

Custom sink — implement EventLogProtocol to send events to Postgres, Kafka, or JSONL:

from praisonaiagents.bus import EventBus

class JsonlSink:
    def __init__(self, path):
        self._f = open(path, "a")

    def append(self, event):
        self._f.write(f"{event.type} {event.data}\n")

    def query(self, session_id, after_seq=0, limit=500):
        return []

    def prune(self, *, older_than_days, max_rows):
        return 0

bus = EventBus()
bus.attach_sink(JsonlSink("events.jsonl"))

Best Practices

Related