Skip to content

Activity Persistence

When an AI agent uses tools during a conversation, the full timeline matters: which tools were called, with what arguments, what they returned, and how long they took. RoomKit persists this interleaving as individual events rather than flattening everything into a single text blob.

The Problem

Without activity persistence, a streaming AI response like:

text: "Let me search for that..."
tool: web_search(q="roomkit docs") -> {results: [...]}
text: "I found 3 results. Let me fetch the page..."
tool: web_fetch(url="...") -> {html: "..."}
text: "Here's what I found: ..."

Would be stored as a single concatenated string, losing tool call boundaries, arguments, results, and timestamps. On reload, the client sees one wall of text with no trace of what the AI actually did.

How It Works

RoomKit persists each segment of an AI response as a separate RoomEvent:

Index Type Content
1 message "Let me search for that..."
2 tool_call_start web_search, args: {q: "roomkit docs"}, status: pending
3 tool_call_end web_search, result: {results: [...]}, status: completed, 1200ms
4 message "I found 3 results. Let me fetch..."
5 tool_call_start web_fetch, args: {url: "..."}, status: pending
6 tool_call_end web_fetch, result: {html: "..."}, status: completed, 800ms
7 message "Here's what I found: ..."

All events share the same correlation_id, linking them as parts of one AI response. Each has its own created_at timestamp and sequential index.

This works identically for both streaming and non-streaming AI providers.

ToolCallContent

Tool call events use the ToolCallContent content type:

from roomkit import ToolCallContent

# At tool start (persisted automatically by the framework)
ToolCallContent(
    tool_name="web_search",
    tool_id="call-abc123",
    arguments={"q": "roomkit docs"},
    status="pending",
)

# At tool end
ToolCallContent(
    tool_name="web_search",
    tool_id="call-abc123",
    arguments={"q": "roomkit docs"},
    result={"results": [...]},
    status="completed",  # or "failed"
    duration_ms=1200,
    error=None,  # populated on failure
)

Querying the Store

Full Timeline

Retrieve everything the room received, in order (a row a hook refused is served on request, see Rows the Room Refused):

timeline = await store.get_timeline(room_id)

for event in timeline:
    if event.type == EventType.MESSAGE:
        print(f"[{event.created_at}] Text: {event.content.body}")
    elif event.type == EventType.TOOL_CALL_START:
        print(f"[{event.created_at}] Tool started: {event.content.tool_name}")
    elif event.type == EventType.TOOL_CALL_END:
        print(f"[{event.created_at}] Tool done: {event.content.tool_name} ({event.content.duration_ms}ms)")

A TOOL_CALL_END states how its call ended in outcome: served, refused (the channel or a gate stopped it, or the handler refused it), failed, blocked (an ON_TOOL_CALL hook withheld the result), unserved (nothing served it) or cancelled. status folds it into completed / failed; a row written before the field existed reads by its status (outcome is None).

Which Way a Page Reads

Every list_events read, and every read built on it, answers two questions separately: which events the page holds, and in what order they come back. newest_first=True and before_index= choose the window -- the room's tail, or the limit events just before a cursor; under newest_first, offset counts back from the newest end. The page itself is always ascending by index: the first element is the oldest event of the window, the last element the newest. Read the most recent event of a page as page[-1], never as page[0]:

# Seven events msg1..msg7 in the room
page = await store.list_events(room_id, limit=5, newest_first=True)
# page -> [msg3, msg4, msg5, msg6, msg7]
latest = page[-1]  # msg7

The same holds for get_timeline(..., newest_first=True) and for get_conversation(), which are built on list_events (RFC §14.1).

Conversation Only (AI Context)

For rebuilding AI context, use get_conversation() which returns only MESSAGE events -- no tool call noise:

messages = await store.get_conversation(room_id, limit=50)
# Only text messages, suitable for feeding back to the AI provider

Without a cursor this returns the most recent limit messages, in ascending order -- messages[-1] is the newest message in the room. This is what fills RoomContext.recent_events, so a room whose history outgrew the limit still hands hooks and AI channels the current conversation, not its opening.

Pass after_index to turn it into a forward cursor instead -- the first limit messages after that index, which is what you want to page through everything that arrived since your last read:

new_messages = await store.get_conversation(room_id, limit=50, after_index=last_seen)

Filtering with EventFilter

EventFilter provides fine-grained control over queries:

from roomkit import EventFilter, EventType

# Get all tool calls in a room
tools = await store.list_events(
    room_id,
    event_filter=EventFilter(
        event_types=[EventType.TOOL_CALL_START, EventType.TOOL_CALL_END],
    ),
)

# Get events from a specific AI response (by correlation_id)
response_events = await store.list_events(
    room_id,
    event_filter=EventFilter(correlation_id="abc123"),
)

# Get events from a specific channel
ai_events = await store.list_events(
    room_id,
    event_filter=EventFilter(source_channel_id="ai-assistant"),
)

# Time-range queries
from datetime import datetime, UTC, timedelta

recent = await store.list_events(
    room_id,
    event_filter=EventFilter(
        after_time=datetime.now(UTC) - timedelta(hours=1),
    ),
)

# Combine filters
recent_tools = await store.list_events(
    room_id,
    event_filter=EventFilter(
        event_types=[EventType.TOOL_CALL_END],
        source_channel_id="ai-assistant",
        after_time=datetime.now(UTC) - timedelta(hours=1),
    ),
)

Rows the Room Refused

An event a BEFORE_BROADCAST hook blocked, or that a read-only or muted source sent, is committed with status=BLOCKED and delivered to nobody. A timeline read skips those rows by default: list_events() and get_timeline() return what the room received, and the filter applies before the page is cut, so a page of limit events is full whatever was refused around them. An audit, copy or deletion reader asks for the refused rows explicitly, and get_event() always answers by id:

from roomkit import EventFilter, EventStatus

received = await store.list_events(room_id)  # no BLOCKED row
everything = await store.list_events(room_id, event_filter=EventFilter(include_blocked=True))
refused = [e for e in everything if e.status == EventStatus.BLOCKED]

The count follows the same rule as the page it stands for (RFC §14.1): get_event_count(room_id) counts every committed row, the refused ones included (a refused row consumed an index), while get_event_count(room_id, event_filter) counts exactly the rows a list_events page would serve under that filter, with no page and no cap:

from roomkit import ChannelType, EventFilter, EventType

everything = await store.get_event_count(room_id)  # refused rows included
ai_turns = await store.get_event_count(
    room_id, EventFilter(event_types=[EventType.MESSAGE], source_channel_type=ChannelType.AI)
)  # received rows only, like the page

get_conversation() is the exception: it fills RoomContext.recent_events, which hooks read whole (RFC §7.5 rule 8), so it returns the refused messages too. Channels never see them: visible_events drops them per reader.

Persistence Policy

Control which event types are persisted with PersistencePolicy:

from roomkit import RoomKit, PersistencePolicy, EventType

# Persist everything except typing and presence indicators
kit = RoomKit(
    persistence_policy=PersistencePolicy(
        exclude_types={EventType.TYPING, EventType.PRESENCE},
    ),
)

# Or whitelist specific types
kit = RoomKit(
    persistence_policy=PersistencePolicy(
        persist_types={
            EventType.MESSAGE,
            EventType.TOOL_CALL_START,
            EventType.TOOL_CALL_END,
        },
    ),
)

exclude_types always takes precedence over persist_types. When persist_types is None (default), all event types are persisted.

The policy is checked before every add_event call in the framework. Events excluded by the policy are still broadcast to channels in real-time -- they just aren't written to the store.

Architecture

Streaming Path

During streaming, the AI generator yields structured markers alongside text deltas:

str("Let me search...")  ->  transport channels (real-time display)
ToolCallStartMarker      ->  triggers text segment persistence
ToolCallEndMarker        ->  triggers tool call end persistence
str("Here are results")  ->  transport channels (real-time display)

Transport channels only see string deltas for real-time rendering. The framework's streaming consumer intercepts markers and persists events at each boundary.

Providers That Do Not Stream

A provider that does not stream takes the same path. The default generate_structured_stream wraps its generate() response into the same events, so its turn yields the same markers and its rows are stored at the same boundaries.

Broadcast Behavior

Tool call events (TOOL_CALL_START, TOOL_CALL_END) are committed and delivered like the turn's other events: every bound channel receives them, streaming transports included (a stream renders text, not tool cards). AI channels ignore them on the way in, so no intelligence channel answers another agent's tool call. On the streaming path they cross the BEFORE_BROADCAST sync hooks before they commit, as the text segments do: a hook can modify or block them, and its tasks, observations and injected events are kept either way.

BEFORE_BROADCAST and AFTER_BROADCAST hooks fire for all event types, so hook authors can shape tool call events before they are stored and observe them for analytics, logging, or auditing.