Store

ConversationStore

Bases: ABC

Persistent storage for rooms, events, bindings, and participants.

Implement this ABC to plug in any storage backend (SQL, Redis, etc.). The library ships with InMemoryStore for development and testing.

create_room abstractmethod async

create_room(room)

Persist a new room.

get_room abstractmethod async

get_room(room_id)

Get a room by ID, or None if it doesn't exist.

update_room abstractmethod async

update_room(room)

Update an existing room.

patch_room_metadata async

patch_room_metadata(room_id, patch, *, unset=())

Merge patch into a room's metadata, removing unset keys first.

The targeted alternative to update_room for metadata-only changes. update_room rewrites the whole room row from an in-memory Room (read-modify-write), so a caller holding a stale object silently clobbers concurrent metadata patches and regresses the counters (event_count / latest_index / timers) advanced by commit_event. This method touches only the metadata keys it is given and stamps updated_at.

Returns the updated room, or None when room_id does not exist.

The default implementation is NOT atomic — it reads, merges, then writes back via update_room. Persistent backends that may be shared across processes MUST override this with a single storage-level partial update (e.g. a JSONB merge).

delete_room abstractmethod async

delete_room(room_id)

Delete a room. Returns True if the room existed.

list_rooms abstractmethod async

list_rooms(offset=0, limit=50)

List rooms with pagination.

find_rooms abstractmethod async

find_rooms(organization_id=None, status=None, metadata_filter=None, *, limit=100, offset=0)

Find rooms matching the given filters.

find_latest_room abstractmethod async

find_latest_room(participant_id, channel_type=None, status=None)

Find the most recent room for a participant.

find_room_id_by_channel abstractmethod async

find_room_id_by_channel(channel_id, status=None)

Find a room ID that has a binding for the given channel_id.

find_room_ids_by_channel async

find_room_ids_by_channel(channel_id, status=None, limit=2)

Room IDs bound to channel_id, deterministically ordered.

Routing must be able to tell "exactly one room" from "several", because picking one of several delivers a message into a conversation it does not belong to (RFC §10.4). find_room_id_by_channel cannot express that — it returns one id either way — so this returns up to limit, and the default of 2 is enough to answer "is this ambiguous?".

Implementations MUST order the result deterministically: the same stored state has to give the same answer whatever the backend.

This is not abstract, so a store written before it existed keeps working — it falls back to the single lookup and therefore cannot report ambiguity. Override it.

add_event abstractmethod async

add_event(event)

Store a new event.

get_event abstractmethod async

get_event(event_id)

Get an event by ID.

update_event abstractmethod async

update_event(event)

Update an existing event (e.g., mark as edited or deleted).

delete_event abstractmethod async

delete_event(room_id, event_id, *, cascade_replies=True)

Hard-delete a persisted event, optionally cascading to its thread replies.

parent_event_id has no DB-level FK, so a bare root delete would orphan its replies — cascade_replies (default) removes them in the same operation. Returns the deleted event IDs (root first, then replies); empty when event_id does not exist in room_id.

list_events abstractmethod async

list_events(room_id, offset=0, limit=50, visibility_filter=None, *, after_index=None, before_index=None, event_filter=None, newest_first=False)

List events in a room with pagination and filtering.

Supports two pagination modes:

• Offset-based (default): offset + limit for simple page access.
• Cursor-based: after_index or before_index for efficient keyset pagination on large rooms. When either is set, offset is ignored.

after_index and before_index are mutually exclusive.

When event_filter is provided, its visibility field takes precedence over visibility_filter.

By default the offset-based mode returns the oldest limit events (the head of the room). Pass newest_first=True to return the most recent limit events instead — still in ascending chronological order, so a "give me the latest page" snapshot reads top-to-bottom. newest_first only applies to the offset-based mode; it is ignored when a cursor (after_index / before_index) is supplied.

.. note::

Cursor pagination relies on events having a valid ``index``
assigned by :meth:`add_event_auto_index`.  Events stored via
:meth:`add_event` keep the model default ``index=0`` and will
all compare equal, producing incorrect cursor results.  Use
``add_event_auto_index`` for rooms that need cursor pagination.

Parameters:

Name	Type	Description	Default
room_id	str	Room to query.	required
offset	int	Number of events to skip (offset-based mode).	0
limit	int	Maximum number of events to return.	50
visibility_filter	str | None	Optional visibility value to filter by. Ignored when event_filter provides a visibility.	None
after_index	int | None	Return events with index > after_index (ascending).	None
before_index	int | None	Return the last limit events with index < before_index, in ascending order.	None
event_filter	EventFilter | None	Rich filter criteria (event types, source, time range, correlation ID). See :class:EventFilter.	None
newest_first	bool	In offset-based mode, return the most recent limit events (ascending order) instead of the oldest. Ignored when a cursor is supplied.	False

get_thread_summaries abstractmethod async

get_thread_summaries(room_id, root_event_ids)

Return reply aggregates for the given thread roots.

For each root that has replies, the result maps its id to a :class:ThreadSummary (reply count + last-reply timestamp). Roots with no replies are absent from the mapping. Used to render a "N replies" affordance without fetching every reply.

check_idempotency abstractmethod async

check_idempotency(room_id, key)

Check if an idempotency key has been seen. Returns True if duplicate.

get_event_count abstractmethod async

get_event_count(room_id)

Return the total number of events in a room.

add_event_auto_index async

add_event_auto_index(room_id, event)

Atomically assign the next index and store the event.

The default implementation reads the count and writes in two steps. Backends should override this with an atomic implementation (e.g. a single SQL transaction) to prevent race conditions on the index.

commit_event async

commit_event(room_id, event)

Store event and bump the room counters as ONE atomic commit.

The commit point of RFC §10.1 (step 12) / §14.3: (re)assigning the authoritative index, persisting the event, and updating the room's event_count / latest_index / timers.last_activity_at form a single logical transaction. An observer MUST never see a stored event that the room counters do not reflect, nor counters that count an event absent from the timeline.

The authoritative index is (re)computed inside the commit (RFC §8.1) so a persistent store shared across processes serializes concurrent writers on the store, not only on an in-process room lock.

Returns the committed event — its index may differ from any provisional value the caller assigned before hooks ran, if the store serialized a concurrent writer.

The default implementation is NOT atomic — it indexes, inserts, then updates the room in separate steps. Persistent backends that may be shared across processes MUST override this with a single storage transaction.

get_conversation async

get_conversation(room_id, *, limit=50, after_index=None)

Return message events only — suitable for AI context rebuilding.

Filters to MESSAGE events, excluding tool calls, lifecycle, and system noise. Tool call history is not included because AI providers track their own tool call context internally via the message history passed to each generation call.

Use :meth:get_timeline to retrieve the full activity log including tool calls.

get_timeline async

get_timeline(room_id, *, event_filter=None, limit=100, after_index=None, newest_first=False)

Return the full activity timeline for a room.

Returns all persisted events in order. Use event_filter to narrow results (e.g. only tool calls, only a specific correlation group).

Without a cursor the default is the oldest limit events; pass newest_first=True for the most recent limit (still ascending) — the right shape for a reconnect snapshot that must show recent history, not the room's opening events.

add_binding abstractmethod async

add_binding(binding)

Attach a channel binding to a room.

get_binding abstractmethod async

get_binding(room_id, channel_id)

Get a channel binding, or None if not attached.

update_binding abstractmethod async

update_binding(binding)

Update an existing channel binding.

remove_binding abstractmethod async

remove_binding(room_id, channel_id)

Detach a channel from a room. Returns True if it was attached.

list_bindings abstractmethod async

list_bindings(room_id)

List all channel bindings for a room.

add_participant abstractmethod async

add_participant(participant)

Add a participant to a room.

get_participant abstractmethod async

get_participant(room_id, participant_id)

Get a participant by ID within a room.

update_participant abstractmethod async

update_participant(participant)

Update a participant.

list_participants abstractmethod async

list_participants(room_id)

List all participants in a room.

create_identity abstractmethod async

create_identity(identity)

Create a new identity record.

get_identity abstractmethod async

get_identity(identity_id)

Get an identity by ID.

resolve_identity abstractmethod async

resolve_identity(channel_type, address)

Look up an identity by channel type and address.

link_address abstractmethod async

link_address(identity_id, channel_type, address)

Link a channel address to an identity.

add_task abstractmethod async

add_task(task)

Store a new task.

get_task abstractmethod async

get_task(task_id)

Get a task by ID.

list_tasks abstractmethod async

list_tasks(room_id, status=None)

List tasks for a room, optionally filtered by status.

update_task abstractmethod async

update_task(task)

Update a task.

add_observation abstractmethod async

add_observation(observation)

Store a new observation.

list_observations abstractmethod async

list_observations(room_id)

List all observations for a room.

mark_read abstractmethod async

mark_read(room_id, channel_id, event_id)

Mark an event as read for a channel.

mark_all_read abstractmethod async

mark_all_read(room_id, channel_id)

Mark all events as read for a channel.

get_unread_count abstractmethod async

get_unread_count(room_id, channel_id)

Return the number of unread events for a channel in a room.

list_read_markers abstractmethod async

list_read_markers(room_id)

Return every channel's read high-water-mark in a room.

Maps channel_id -> the highest read event index. Channels with no marker are absent. With one channel per member, this is the raw material for aggregating per-member "seen by" receipts.

close async

close()

Release any resources held by the store (e.g. a connection pool).

Called by RoomKit.close(). The default is a no-op — override it in backends that own external resources. Implementations MUST be idempotent and MUST NOT close resources they do not own.

InMemoryStore

InMemoryStore()

Bases: ConversationStore

Dict-based in-memory store for development and testing.

add_event_auto_index async

add_event_auto_index(room_id, event)

Atomically assign index = len(room_events) and append.

commit_event async

commit_event(room_id, event)

Atomic commit (RFC §10.1 step 12 / §14.3): index, store, and bump the room counters under the per-room lock as one critical section, so the timeline and the counters can never diverge.

PostgresStore

PostgresStore(dsn=None, pool=None)

Bases: ConversationStore

PostgreSQL-backed conversation store using asyncpg.

Uses a fully relational schema with indexed columns for all frequently queried fields. JSONB is used only for flexible/extensible data (metadata, content, capabilities).

init async

init(min_size=2, max_size=10)

Create the connection pool (if needed) and ensure the schema exists.

Runs only additive, idempotent DDL (CREATE TABLE IF NOT EXISTS). It never drops a table, so calling init() after a library upgrade cannot destroy data.

If a v1 (JSONB-blob) schema is detected, init() refuses to touch it and raises :class:PostgresSchemaError. Back up your data, then run the explicit :meth:migrate to move v1 → v2.

migrate async

migrate(*, dry_run=True, confirm=False)

Explicit, opt-in schema migration. Never runs automatically.

The only path that performs the destructive v1 → v2 migration, which DROPs every RoomKit table (irreversible data loss). Serialized across processes with a PostgreSQL advisory lock.

Parameters:

Name	Type	Description	Default
dry_run	bool	When True (the default) nothing is executed — the returned report says what would happen. Set False to run.	True
confirm	bool	Required together with dry_run=False to run the destructive drop. It is your acknowledgement that you have a backup; without it the call raises rather than deleting data.	False

Returns:

Type	Description
dict[str, Any]	A report {"detected_version", "action", "dropped_tables"} where
dict[str, Any]	action is one of "noop" (already v2), "dry_run", or
dict[str, Any]	"migrated".

Raises:

Type	Description
PostgresSchemaError	on dry_run=False without confirm=True.

dedupe_event_indices async

dedupe_event_indices(*, dry_run=True)

Repair duplicate event indices, then enforce UNIQUE(room_id, index).

A pre-fix release could assign the same index to two events in a room under concurrency (RFC §8.1). This renumbers each affected room's events to a unique, sequential 0..N-1 ordered by (index, created_at, id), reconciles the room counters, and (re)creates idx_events_room_index as UNIQUE — all in one transaction.

Renumbering shifts indices, so read markers (last_read_index / read_markers.event_index) may be off afterwards. Run it in a maintenance window, on a backup first.

Parameters:

Name	Type	Description	Default
dry_run	bool	When True (the default) nothing is changed — the report only counts what would be repaired.	True

Returns:

Type	Description
dict[str, Any]	{"action", "duplicate_rows", "affected_rooms", "now_unique"}
dict[str, Any]	where action is "dry_run", "noop" (already clean), or
dict[str, Any]	"repaired".

drop_legacy_parent_index async

drop_legacy_parent_index(*, dry_run=True)

Drop the pre-composite single-column idx_events_parent.

idx_events_parent_index on (parent_event_id, index) — created additively by :meth:init — supersedes it: the leading column still serves plain parent_event_id lookups while the second lets thread pagination read the page pre-sorted. :meth:init is additive and never drops, so this opt-in call removes the now-redundant single-column index on databases that predate the composite. Idempotent — a no-op once gone.

Parameters:

Name	Type	Description	Default
dry_run	bool	When True (the default) nothing is changed; the report only says whether the legacy index is still present.	True

Returns:

Type	Description
dict[str, Any]	{"action"} — "dry_run", "noop" (already gone), or
dict[str, Any]	"dropped".

close async

close()

Release the connection pool if we own it.

patch_room_metadata async

patch_room_metadata(room_id, patch, *, unset=())

Atomic override: one partial UPDATE, no read-modify-write.

(metadata - unset) || patch runs inside the row update itself, so concurrent patches merge instead of clobbering and the counters maintained by commit_event are never touched.

add_event_auto_index async

add_event_auto_index(room_id, event)

Atomically assign the next index and store the event in one transaction.

Uses SELECT ... FOR UPDATE on the rooms table to serialise concurrent index assignments for the same room.

commit_event async

commit_event(room_id, event)

Atomic commit (RFC §10.1 step 12 / §8.1 / §14.3) in ONE transaction.

SELECT ... FOR UPDATE on the room row serializes concurrent commits for the same room across connections (i.e. across processes); the index is computed, the event inserted, and the room counters (event_count / latest_index / timers.last_activity_at) updated together, so the timeline and the counters can never diverge — even under a crash or without a cross-process room lock.