Session Mirror

Mirror session transcripts to a remote backend without blocking the local turn

A mirror replicates each persisted turn to a remote backend on a background thread, so a session can follow the user across machines — local disk stays the source of truth.

Quick Start

1. Plug a mirror into DefaultSessionStore

   from typing import Any, Dict, List
   
   from praisonaiagents import Agent
   from praisonaiagents.session import DefaultSessionStore, SessionMirrorProtocol
   
   
   class LoggingMirror:
       """Minimal SessionMirrorProtocol implementation — keeps records in memory."""
   
       def __init__(self):
           self._store: Dict[str, List[Dict[str, Any]]] = {}
   
       def append(self, session_id: str, records: List[Dict[str, Any]]) -> None:
           self._store.setdefault(session_id, []).extend(records)
           print(f"[mirror] +{len(records)} record(s) for {session_id}")
   
       def load(self, session_id: str) -> List[Dict[str, Any]]:
           return list(self._store.get(session_id, []))
   
   
   store = DefaultSessionStore(mirror=LoggingMirror())
   agent = Agent(
       name="Assistant",
       instructions="Be concise.",
       memory={"session_id": "user-42", "session_store": store},
   )
   agent.start("Remember I like green tea.")

   The local file at ~/.praisonai/sessions/user-42.json and the mirror both receive the turn.

2. Continue on another machine

   store = DefaultSessionStore(mirror=LoggingMirror())
   records = store._mirror.load("user-42")  # hydrate a session never seen locally

   A configured mirror’s load(session_id) returns the full transcript, so a fresh install can pick up the conversation.

How It Works

The store writes local disk first and returns; the record is then handed to a bounded queue drained by a daemon thread that calls mirror.append.

The failure modes never touch the local write:

Configuration Options

SessionMirrorProtocol methods

The runtime_checkable protocol lives at praisonaiagents.session.SessionMirrorProtocol.

DefaultSessionStore(mirror=…)

Writing a Mirror

A mirror is three tiny methods — the LoggingMirror from Quick Start is the whole story.

from typing import Any, Dict, List


class LoggingMirror:
    def __init__(self):
        self._store: Dict[str, List[Dict[str, Any]]] = {}

    def append(self, session_id: str, records: List[Dict[str, Any]]) -> None:
        self._store.setdefault(session_id, []).extend(records)

    def load(self, session_id: str) -> List[Dict[str, Any]]:
        return list(self._store.get(session_id, []))

Local-First Guarantees

Testing & Reconciliation

Use flush_mirror(timeout=…) to make mirroring deterministic in a test and close_mirror() to drain in a teardown.

store = DefaultSessionStore(mirror=my_mirror)
# ... run turns ...
assert store.flush_mirror(timeout=2.0), "mirror did not drain in time"
store.close_mirror()   # stops the daemon writer after draining

Backward Compatibility

mirror=None is the default — existing DefaultSessionStore(...) call sites are byte-for-byte unchanged, with no daemon thread spawned and no dependency added.

Best Practices

Related