How to Persist Conversations and State

Picture the production shape of an agent deployment: several replicas of the API server behind a load balancer, and clients that come and go. A browser tab starts a turn, closes, reopens; a phone picks up the same conversation; a replica gets recycled by a deploy mid-answer.

The SDK closes each of those gaps with a storage interface, and leaves the production backend to you: implement the interface on whatever your infrastructure provides (an object store, a SQL database, Redis) and pass it to setup_app. File-backed implementations of the state, recording, supervision, and MCP OAuth stores ship with the SDK for local development and single-node deployments; the conversation store ships as an interface only. This page walks the problems first, then the wiring, then the contracts your adapters must honor.

Making a dependency resumable covers the producer side of checkpointing: how a dependency serializes its working state. This page covers where that state goes, and everything else the runtime persists.

The Problems

Every problem below plays out in the same scene: clients reach the replicas through a load balancer, and whatever a turn produces (the live stream, the agent's working state) exists only in the memory of the replica running it.

A reconnect lands on the wrong replica

The client disconnects (tab close, network blip) and reconnects through the load balancer, which routes it to any replica. The in-memory stream buffer exists only on the replica producing the turn, so without shared storage the reconnect finds nothing, even though the turn is still running fine.

The stream recording closes this gap: the producer keeps flushing the latest cumulative message of the in-flight turn to shared storage, and any replica can serve the reconnect by following that record, pacing the content the client has not seen out as a smooth stream.

The same turn on a second screen

You start a long turn on your laptop and open the conversation on your phone. Both clients should watch the same answer grow, and a client joining mid-turn should catch up from the beginning of the message. On the producing replica this is the in-memory buffer fanning out to any number of consumers; everywhere else it is the same recording-following path as a reconnect, so every device gets the stream no matter where its connection lands. Discovery is part of the problem: a client landing on a conversation first asks GET /chat/stream/status which turn is in flight, then attaches to it.

The producer dies mid-turn

A deploy, a crash, or a lost node kills the replica producing the turn. An agent twenty minutes into searching, reading, and writing must not lose that work.

This is what the agent state store exists for: during a stateful turn, a background worker keeps checkpointing the working state of every resumable dependency. When the producer dies, a resume request revives the agent on another replica from the last checkpoint: completed tool calls, retrieved context, and generated text are restored rather than recomputed, and the model/tool loop re-drives from where it left off.

Knowing what is true right now

Connects, reconnects, and takeovers all hinge on meta-information: is there a live producer for this conversation, and which turn does it own? A reconnecting client needs different treatment depending on that: follow the recording if the producer is alive, revive the agent from its checkpoint if it is dead, refetch the conversation if the turn already finished. And two producers must never run the same conversation at once.

The supervision store holds that meta-information: one small row per conversation with the owning turn, a heartbeat, and a lease epoch. Liveness is always decided by the reader from the heartbeat's age; a crashed process cannot record its own death. The row doubles as the conversation's mutex (starting a turn while a fresh one is running is refused with a 409 carrying the occupant) and as the ownership lease: a takeover claims a higher epoch, and every producer write is epoch-guarded, so a producer that was only slow, not dead, fences itself the moment its writes start failing.

Conversations outlive their turns

The mundane half of the problem: users list their conversations and reopen them later, so the transcript must be durable, keyed by conversation, and owned by its user. The conversation store holds that transcript, and the chat handler interacts with it at precise points:

• The incoming user message is appended before the agent runs, so the user's turn survives a producer crash; the completed assistant message is appended after the turn finishes. A turn that errors moves the conversation's last-turn status to errored instead of leaving it permanently running.
• Because the user turn is already stored, a resume request does not re-record it, and a client never needs to resend history: on stateful follow-ups the handler feeds agents that do not replay their own transcript the stored one instead of the request body. That is what lets clients send only the new message.
• After each completed turn the handler emits a ConversationUpdated event (ids only) on the message bus, the hook for mirroring transcripts into other systems.

Writes are owner-only; reads are allowed for shared conversations.

Acting on the user's behalf on MCP servers

An MCP server that requires OAuth cannot be authorized inline: the OAuth2 authorization-code dance ends with a browser callback, and that callback can land on any replica, minutes after the flow started. So the in-flight state must be persisted (the state parameter, the PKCE verifier, the token endpoint, the redirect URI, an expiry) for the callback to finalize the exchange, and the granted tokens must be stored per (tenant, user, server) so the agent can act within the scopes the user granted, across turns and replicas. The MCP OAuth store holds both.

How the Stores Answer Them

Problem	Store	Without it
Reconnect on any replica, second screens	ChatStreamRecordingStore	Reattachment works only on the producing replica, while its buffer lives
Producer death without losing work	ChatAgentStateStore	Stateful requests fail with 400 / AGENTS_STORAGE_NOT_CONFIGURED; no resume
Liveness, ownership, one turn per conversation	ChatStreamSupervisionStore	No discovery, no 409 mutex, no dead-producer verdicts, no fencing
Durable, listable conversations; clients send only the new turn	ChatConversationStore	Only agents that restore their own transcript can run stateful follow-ups
OAuth on behalf of the user	MCPOAuthRepository, wrapped by MCPOAuthIntegrationStore	MCP servers requiring user authorization are skipped

Each store is optional and unlocks its own layer: start with the state store and add the others as you need them.

Wiring the Stores

setup_app accepts one argument per store. The bundled file-backed implementations use atomic replace and file locks, so they are correct for any number of server processes sharing one directory: fine for local work and single-node deployments, and what the crash-test harness runs on.

import os

from zav.agents_sdk import AgentSetupRetrieverFromFile, setup_app
from zav.agents_sdk.adapters import AgentDependencyRegistry
from zav.agents_sdk.adapters.agent_state import LocalFileChatAgentStateStore
from zav.agents_sdk.adapters.local_agent_registries_factory import (
    LocalAgentRegistriesFactory,
)
from zav.agents_sdk.adapters.stream_buffer import (
    LocalFileChatStreamRecordingStore,
    LocalFileChatStreamSupervisionStore,
)
from zav.agents_sdk.cli.load_chat_agent_factory import (
    from_string as import_chat_agent_class_registry_from_string,
)

storage_path = os.path.abspath(".agent-state")

chat_agent_class_registry = import_chat_agent_class_registry_from_string("./agents")
agent_setup_retriever = AgentSetupRetrieverFromFile(
    file_path="./agents/agent_setups.json",
    secret_file_path="./agents/env/agent_setups.json",
)

app = setup_app(
    agent_registries_factory=LocalAgentRegistriesFactory(
        agent_setup_retriever=agent_setup_retriever,
        chat_agent_class_registry=chat_agent_class_registry,
        agent_dependency_registry=AgentDependencyRegistry,
    ),
    agent_state_store=LocalFileChatAgentStateStore(
        base_path=os.path.join(storage_path, "state")
    ),
    chat_stream_recording_store=LocalFileChatStreamRecordingStore(
        base_path=os.path.join(storage_path, "recording")
    ),
    chat_stream_supervision_store=LocalFileChatStreamSupervisionStore(
        base_path=os.path.join(storage_path, "supervision")
    ),
)

The example wires the three stores that ship with file-backed implementations. The other two follow the same pattern: pass your ChatConversationStore adapter as chat_conversation_store, and the MCP OAuth pair as shown in its section below.

For a multi-replica deployment, replace the file-backed classes with your own adapters, and point every replica at the same backend: a turn checkpointed by one replica is resumed by another, and a stream produced on one replica is followed from another, only if they read the same rows.

Implementing an Adapter

Subclass the abstract interface and implement its methods. The four store interfaces are importable from zav.agents_sdk (top level) and the MCP OAuth repository from zav.agents_sdk.adapters.mcp; every store method takes the tenancy triple (tenant, requester_uuid or user_uuid, session_id) so a backend can partition however it likes.

ChatAgentStateStore

Where a dead producer's work survives: the checkpoint blobs a resumed turn is revived from.

from zav.agents_sdk import ChatAgentStateStore

DependencyState = Dict[str, Dict[str, Any]]  # {state_key: blob}

class MyStateStore(ChatAgentStateStore):
    async def load(self, tenant, user_uuid, session_id) -> DependencyState: ...
    async def save(self, tenant, user_uuid, session_id, state) -> None: ...
    async def delete(self, tenant, user_uuid, session_id) -> None: ...

• load returns {} when nothing is saved; delete is a no-op when nothing exists.
• save is called every 5 seconds during a stateful turn and once when the turn finishes. Each call replaces the whole blob, so a plain overwrite is correct.
• construct_agent_state_key(tenant, user_uuid, session_id) (importable from zav.agents_sdk.domain.chat_agent_state_store) builds the canonical relative key, agent-state/{tenant}/{user_uuid}/{session_id}, if your backend wants the same layout as the file store.

ChatStreamSupervisionStore

The liveness record, send mutex, and ownership lease, in one row per conversation. This is the only interface with a hard atomicity requirement.

from zav.agents_sdk import ChatStreamSupervisionStore
from zav.agents_sdk.domain.chat_stream_supervision_store import (
    ChatStreamStatus,      # RUNNING | DONE | ERRORED
    ChatStreamSupervision, # status, heartbeat_at, message_id, epoch
)

class MySupervisionStore(ChatStreamSupervisionStore):
    async def compare_and_swap(
        self, tenant, requester_uuid, session_id,
        expected_epoch,  # Optional[int]; None means "only if no row exists"
        supervision,     # ChatStreamSupervision to write
    ) -> bool: ...
    async def read(self, tenant, requester_uuid, session_id): ...
    async def delete(self, tenant, requester_uuid, session_id, message_id, epoch) -> None: ...

• compare_and_swap must atomically write supervision if and only if the current row's epoch equals expected_epoch (or no row exists and expected_epoch is None), returning whether it wrote. Use a serializable transaction, a conditional write, or an equivalent primitive; a read-then-write without atomicity lets two replicas both claim a conversation. The file store uses an exclusive flock for this.
• delete removes the row only if it still holds exactly message_id at epoch, atomically with that check. A row already claimed by a newer turn is left alone.
• There is no stored "dead" status: the owner stamps a fresh RUNNING heartbeat every second, and the framework reads a heartbeat_at more than 3 seconds old as a dead producer.

ChatStreamRecordingStore

The snapshot a follower streams a reconnect from.

from zav.agents_sdk import ChatStreamRecordingStore
from zav.agents_sdk.domain.chat_stream_recording_store import RecordedStreamEvent
# RecordedStreamEvent: index, message (ChatMessage), message_id, epoch

class MyRecordingStore(ChatStreamRecordingStore):
    async def put(self, tenant, requester_uuid, session_id, event) -> None: ...
    async def read(self, tenant, requester_uuid, session_id): ...
    async def delete(self, tenant, requester_uuid, session_id, message_id) -> None: ...

• One overwritten record per conversation, not an append log: every streamed message is cumulative (it contains the whole answer so far), so keeping only the newest is loss-free. The producer flushes at most every 0.5 seconds; a follower diffs by index to serve only what its client has not seen.
• put must reject a write whose epoch is below the stored record's, atomically with reading it, so a fenced producer cannot overwrite its successor's snapshot. delete removes the record only if it still belongs to message_id.
• Writes must be atomic at the record level (no torn reads for a concurrent reader). An object-store blob overwrite or a single-row upsert both qualify.

ChatConversationStore

The durable transcript.

from zav.agents_sdk import ChatConversationStore
# LastTurnStatus = "running" | "complete" | "errored" | "cancelled"

class MyConversationStore(ChatConversationStore):
    async def append_messages(
        self, tenant, requester_uuid, session_id, agent_identifier, messages,
        user_roles=None, user_tenants=None, user_agent_id=None, bot_params=None,
        last_turn_status=None, create_if_absent=True, conversation_context=None,
    ) -> None: ...
    async def set_last_turn_status(self, tenant, requester_uuid, session_id, last_turn_status) -> None: ...
    async def get_transcript(self, tenant, requester_uuid, session_id): ...

• A conversation is keyed by (tenant, session_id) with requester_uuid as its owner (missing normalizes to ""). Writes are owner-only: raise ConversationNotOwnedError (from zav.agents_sdk.domain.chat_conversation_store) for any other requester and the server maps it to 403. Reads are allowed for shared conversations.
• The handler calls append_messages for the user turn with create_if_absent=True and for the bot turn with create_if_absent=False, so a conversation deleted mid-stream is not resurrected as a ghost row.
• get_transcript returns [] when no row exists.
• No file-backed implementation ships for this interface; to use it you implement an adapter (a single-table SQL backend is the natural fit).

MCP OAuth Storage

The persisted OAuth dance and its outcome, split across two classes: MCPOAuthRepository is the storage interface you implement, plain CRUD over the records below; MCPOAuthIntegrationStore wraps a repository and owns the flow rules (expiring stale pending authorizations, clearing a pending entry when tokens land), keeping adapters storage-only. You wire the wrapper, together with the token client that performs the code-for-token exchange:

from zav.agents_sdk import (
    MCPOAuthIntegrationStore,
    get_local_file_mcp_oauth_integration_store,
    get_mcp_oauth_token_client,
)

app = setup_app(
    ...,
    mcp_oauth_store=get_local_file_mcp_oauth_integration_store(),  # or MCPOAuthIntegrationStore(MyOAuthRepository())
    mcp_oauth_token_client=get_mcp_oauth_token_client(),
)

MCPOAuthRepository

from zav.agents_sdk.adapters.mcp import (
    MCPOAuthClientInfoRecord,       # tenant, server_name, client_info
    MCPOAuthPendingAuthorization,   # state, tenant, server_name, user_uuid, auth_url,
                                    # code_verifier, token_endpoint, redirect_uri,
                                    # expires_at, resource
    MCPOAuthRepository,
    MCPUserIntegrationRecord,       # tenant, user_uuid, server_name, pending, tokens
)

class MyOAuthRepository(MCPOAuthRepository):
    async def list_user_integrations(self, tenant, user_uuid, server_names): ...
    async def get_user_integration(self, tenant, user_uuid, server_name): ...
    async def find_user_integration_by_pending_state(self, state): ...
    async def save_user_integration(self, record) -> None: ...
    async def delete_user_integration(self, tenant, user_uuid, server_name) -> bool: ...
    async def list_client_info(self, tenant, server_names): ...
    async def get_client_info(self, tenant, server_name): ...
    async def save_client_info(self, record) -> None: ...

• One MCPUserIntegrationRecord per (tenant, user_uuid, server_name): pending holds an in-flight authorization, tokens the granted OAuth tokens; either can be None. MCPOAuthClientInfoRecord holds per-tenant OAuth client info for servers that use dynamic client registration, keyed by (tenant, server_name).
• find_user_integration_by_pending_state is the callback's entry point and must look up by state alone, across all users and servers: the redirect carries nothing else.
• A pending authorization carries everything needed to finish the exchange plus expires_at; the wrapping store treats expired pendings as gone and cleans them up on read, so the adapter needs no TTL logic of its own.
• Plain last-write-wins CRUD is sufficient; there is no compare-and-swap requirement here.
• Records hold credentials, so a production adapter should encrypt at rest.

Wiring the store also enables the GET /mcp/oauth/callback endpoint that completes the flow.

Validating a Deployment

za agents perf exists to prove this wiring end to end: it runs two real API server processes against one shared store set and a terminal client that starts a long turn, kills a server mid-answer, and reattaches or resumes on the other. See the CLI reference. If your adapters pass a kill-and-resume round trip there, the contracts above are implemented correctly.

Related

• Making Dependencies Resumable - the producer side: how a dependency defines what gets checkpointed.
• Conversations and Streaming - the client-facing behavior these stores enable.
• Extending the API Server - the rest of the setup_app surface.