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.
connection
async
¶
Serve the store calls in this block from one backend connection.
A pooled backend MAY bind a single connection for the duration of the
block, so a stretch of calls pays one checkout instead of one per call.
That checkout is not free: asyncpg, for one, resets a connection on
release (pg_advisory_unlock_all(); CLOSE ALL; UNLISTEN *;
RESET ALL;), a full round trip whose cost rivals the reads it brackets.
The default binds nothing and yields: a store with no connection pool has none to bind, and behaves exactly as it did without the block.
This is not a transaction. No atomicity, no isolation, no rollback,
no snapshot — a failure midway leaves the earlier calls applied, exactly
as it would outside the block. It bounds connection tenure, nothing
else. A caller needing atomicity asks for it explicitly (see
:meth:commit_event).
The block MUST contain store calls and nothing else, awaited
sequentially: no hook, no provider call, no lock acquisition, no
gather/create_task over store calls. Two reasons, both learned
the hard way — holding a pooled connection across foreign code is how a
fleet parks its backends behind someone else's HTTP request, and a child
task inherits this context, so it would use the bound connection
concurrently with its parent.
Reentrant: a nested block joins the outer one rather than taking a second connection.
Yields nothing on purpose. The bound connection is the backend's business; callers keep calling the store's ordinary methods.
update_room
abstractmethod
async
¶
Update an existing room.
Room.delivered_index is store-managed (see
:meth:advance_delivered_index): implementations SHOULD NOT let a
caller's stale copy rewind it. The shipped stores exclude it from
this write path.
room_exists
async
¶
Whether a room exists, without materialising it.
Inbound routing asks this on every message and discards everything
else, so going through :meth:get_room selects every column, decodes
the JSONB ones and validates a whole Room — to answer a yes/no.
Backends SHOULD override with an existence query.
binding_exists
async
¶
Whether a channel is attached to a room, without materialising it.
Same reasoning as :meth:room_exists: routing only needs the yes/no,
and the decision it feeds is re-taken under the room lock anyway.
get_delivered_index
async
¶
Read a room's delivery cursor, and nothing else.
The delivery lane consults this on every turn — twice per event on the
Postgres path, once outside the claim and once under it. Going through
:meth:get_room for it means selecting every column and rebuilding the
whole Room model, JSONB fields decoded and validated, to reach one
integer. Backends SHOULD override this with a single-column read.
Returns -1 for an unknown room, matching the cursor's initial value.
advance_delivered_index
async
¶
Advance the room's delivered index to index (RFC §10.1 step 14).
Compare-and-set: succeeds only from index - 1 — the strict
per-room delivery order — unless force, the delivery-gap skip
policy, which advances from any lower value. Returns True if the
cursor moved.
This default is read-check-write through :meth:get_room /
:meth:update_room; it is only correct under the room's delivery
claim, and only for stores whose update_room persists
delivered_index as given. A store that shields the field there
(as the shipped stores do) MUST override this with an atomic
conditional write.
patch_room_metadata
async
¶
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 a room. Returns True if the room existed.
find_rooms
abstractmethod
async
¶
Find rooms matching the given filters.
find_latest_room
abstractmethod
async
¶
Find the most recent room for a participant.
find_room_id_by_channel
abstractmethod
async
¶
Find a room ID that has a binding for the given channel_id.
find_room_ids_by_channel
async
¶
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.
update_event
abstractmethod
async
¶
Update an existing event (e.g., mark as edited or deleted).
delete_event
abstractmethod
async
¶
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+limitfor simple page access. - Cursor-based:
after_indexorbefore_indexfor efficient keyset pagination on large rooms. When either is set,offsetis 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 |
None
|
before_index
|
int | None
|
Return the last |
None
|
event_filter
|
EventFilter | None
|
Rich filter criteria (event types, source, time range,
correlation ID). See :class: |
None
|
newest_first
|
bool
|
In offset-based mode, return the most recent |
False
|
get_thread_summaries
abstractmethod
async
¶
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 if an idempotency key has been seen. Returns True if duplicate.
get_event_by_idempotency_key
async
¶
The event a previously-seen idempotency key committed (RFC §13.4).
A redelivered message must observe what its first delivery did, not a refusal — a webhook that retries because it never saw the first response is asking what happened, and "duplicate" does not answer it.
Not abstract, and None is a legitimate answer: a store that cannot
resolve the key falls back to reporting the duplicate as blocked, which
is what every store did before this existed.
get_event_count
abstractmethod
async
¶
Count the events a room currently holds, exactly, on demand.
This is the authoritative count and the only one that survives
deletions. Room.event_count is a maintained tally incremented at
each commit (RFC §10.1 step 12) and never decremented, so the two
diverge once events are deleted — deliberately: an exact count is a
scan of the room's whole timeline, and no commit should pay for one.
Call this when the number has to be right. Read Room.event_count
when a cheap, monotonically-growing hint is enough.
add_event_auto_index
async
¶
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
¶
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 the room counters do not reflect.
The converse does not hold, and that is deliberate. event_count is
the running tally RFC §10.1 step 12 describes — event_count += 1,
never a recount — and deletion adjusts no counter (the RFC prescribes
none), so after events are deleted the tally overstates the timeline.
Keeping it exact would mean scanning every event on every commit, which
makes the hot path cost grow with a room's history. Callers who need
the real number call :meth:get_event_count, which is exact and
separate on purpose.
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
¶
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.
Without a cursor this returns the most recent limit messages,
in ascending order — the last element is the newest message in the
room. That is what "the conversation" means for a context rebuild:
the head of a room whose history outgrew limit is history no
model should be answering from. Callers who do want the opening
messages ask for them explicitly via :meth:list_events /
:meth:get_timeline.
With after_index it stays a forward cursor — the first limit
messages after that index, ascending — so keyset pagination
("give me what arrived since I last looked") reads in order and
resumes where it stopped.
The conversation is the host's whole one: a message stored
BLOCKED (refused by a hook, or by a source that could not write)
is in it, as the audit record it is. What a channel may be handed
as history is a per-reader question, and
:func:~roomkit.core.visibility.visible_events answers it (RFC §7.5
rule 8): it drops BLOCKED events along with what visibility withholds.
Returned events are immutable snapshots (RFC §14.4): treat them as frozen. A store may share objects between reads or return fresh ones — rely on neither.
get_timeline
async
¶
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.
get_binding
abstractmethod
async
¶
Get a channel binding, or None if not attached.
remove_binding
abstractmethod
async
¶
Detach a channel from a room. Returns True if it was attached.
get_participant
abstractmethod
async
¶
Get a participant by ID within a room.
list_participants
abstractmethod
async
¶
List all participants in a room.
load_room_context
async
¶
Read a room together with its bindings and participants.
These three reads always travel together when a room's context is
assembled, and that happens on every inbound message. The default
implementation issues them as three separate calls, inside
:meth:connection so a pooled backend that binds one there already
pays a single checkout for all three; overriding this method too buys
the queries themselves in one place, which is what PostgresStore
does.
This is a convenience, NOT a consistent snapshot: the reads carry no more cross-read atomicity than calling the three methods in sequence would. A backend needing that must take it explicitly.
Returns (None, [], []) for a room that does not exist, mirroring
:meth:get_room rather than raising.
resolve_identity
abstractmethod
async
¶
Look up an identity by channel type and address (RFC §14.1).
organization_id scopes the lookup. An address is unique within an organization, not globally: without scoping, a phone number registered by one tenant would resolve to that tenant's identity for every other tenant too (RFC §17.2). Leave it unset in a single-tenant deployment — those registrations live in their own unscoped space.
link_address
abstractmethod
async
¶
Link a channel address to an identity, within an organization.
The same address may be linked to a different identity in each organization: a number belongs to one person at one tenant and to someone else at another.
list_tasks
abstractmethod
async
¶
List tasks for a room, optionally filtered by status.
list_observations
abstractmethod
async
¶
List all observations for a room.
mark_read
abstractmethod
async
¶
Mark an event as read for a channel.
mark_all_read
abstractmethod
async
¶
Mark all events as read for a channel.
get_unread_count
abstractmethod
async
¶
Return the number of unread events for a channel in a room.
list_read_markers
abstractmethod
async
¶
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
¶
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 ¶
Bases: ConversationStore
Dict-based in-memory store for development and testing.
SQLiteStore ¶
Bases: ConversationStore
Single-file persistent store backed by stdlib sqlite3 + FTS5.
search_events
async
¶
Full-text search over stored event text (FTS5, relevance-ranked).
Not part of the :class:ConversationStore contract — an SQLite
extra. query is free text: it is tokenised and matched as an AND
of terms, so user input can be passed through verbatim.
SQLiteSchemaError ¶
Bases: RuntimeError
Raised when a SQLite file cannot be migrated or is from a newer schema.
PostgresStore ¶
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).
connection
async
¶
Run this block's queries on ONE pooled connection.
Implements :meth:ConversationStore.connection — read its contract
before using it, in particular that the block takes store calls only,
sequentially. Every query in PostgresStore goes through
:meth:_acquire, so binding there covers them all without any of them
knowing about it.
One checkout for the whole block instead of one per call, and asyncpg emits its connection reset once instead of once per call.
init
async
¶
Create the connection pool (if needed) and ensure the schema exists.
Runs only additive, idempotent schema maintenance: DDL such as
CREATE TABLE IF NOT EXISTS plus bounded repairs for columns added
by newer versions. It never drops a table or discards user data, so
calling init() after a library upgrade cannot destroy data.
Serialized across processes by the same advisory lock :meth:migrate
takes, acquired as the first statement of the one transaction that
runs the DDL. Idempotent DDL is not the same as concurrent DDL —
two workers resolving IF NOT EXISTS against the same missing
object would each try to create it and the loser would raise — so the
lock makes them run one after the other: the second worker blocks
until the first commits, then finds every object already present and
does nothing. A deploy that restarts a whole fleet at once is exactly
the case this covers.
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
¶
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
|
confirm
|
bool
|
Required together with |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A report |
dict[str, Any]
|
|
dict[str, Any]
|
|
Raises:
| Type | Description |
|---|---|
PostgresSchemaError
|
on |
dedupe_event_indices
async
¶
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
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
|
dict[str, Any]
|
where |
dict[str, Any]
|
|
drop_legacy_parent_index
async
¶
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
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
|
dict[str, Any]
|
|
binding_exists
async
¶
Existence, not the row — routing asks this per message.
get_delivered_index
async
¶
One column, not the whole room.
The lane reads this twice per event — once outside the claim to check whether it is its turn, once under it. Through get_room() that is two full-row SELECTs and two Room models rebuilt per event, JSONB metadata and timers decoded and validated, all to reach a single integer.
advance_delivered_index
async
¶
Atomic CAS on the delivery-lane cursor (RFC §10.1 step 14).
delivered_index is deliberately absent from :meth:update_room's
column list — only this conditional write moves it, so a stale Room
copy can never rewind the lane.
patch_room_metadata
async
¶
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
¶
Atomically assign the next index and store the event in one transaction.
A per-room high-water mark serialises concurrent assignments and is never decremented when events are deleted.
commit_event
async
¶
Atomic commit (RFC §10.1 step 12 / §8.1 / §14.3) in ONE transaction.
SELECT ... FOR UPDATE on the room row serializes counter updates
for the same room across connections (i.e. across processes); a
per-room high-water mark reserves the index, then the event and 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.
load_room_context
async
¶
The three room-scoped reads over ONE pooled connection.
Same queries as the default implementation, one checkout instead of
three. asyncpg resets a connection on release
(pg_advisory_unlock_all(); CLOSE ALL; UNLISTEN *; RESET ALL;), so
each checkout costs a full extra round trip — on the inbound path that
overhead is the same order as the reads it wraps.