Skip to content

SessionStore API

Conversation log for resumable @channel(durable=True) sessions. Backed by the app's StoragePlugin (in-memory, Redis, or SQLAlchemy).

Details: Channels & Agent Sessions.

Durable conversation log for resumable @channel sessions.

Keyed by conversation id over the app's StoragePlugin (InMemory / Redis / SQL), so a session survives a disconnect: the handler reloads history on reconnect and continues where it left off.

SessionStore

Source code in src/istos/primitives/session_store.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class SessionStore:
    def __init__(self, storage: Any) -> None:
        self._storage = storage

    @staticmethod
    def _key(conversation_id: str) -> str:
        return f"conv:{conversation_id}"

    async def append(self, conversation_id: str, direction: str, data: Any) -> None:
        await self._storage.log(
            self._key(conversation_id),
            {"dir": direction, "data": data, "ts": time.time()},
        )

    async def history(self, conversation_id: str, limit: int = 1000) -> List[dict]:
        """Prior messages oldest-first: ``[{dir, data, ts}, ...]``."""
        entries = await self._storage.get_log(self._key(conversation_id), limit=limit)
        return [e["value"] for e in reversed(entries)]

history(conversation_id, limit=1000) async

Prior messages oldest-first: [{dir, data, ts}, ...].

Source code in src/istos/primitives/session_store.py 
26
27
28
29
async def history(self, conversation_id: str, limit: int = 1000) -> List[dict]:
    """Prior messages oldest-first: ``[{dir, data, ts}, ...]``."""
    entries = await self._storage.get_log(self._key(conversation_id), limit=limit)
    return [e["value"] for e in reversed(entries)]