Custom Storage

Implement your own persistence - or put an existing system behind the agent-server API. There are two ways:

1. Extend BaseStorageProvider and implement the _-hooks (it adds the "is connected" check, id generation, validation and the normalisation every provider shares), or
2. implement the StorageProvider interface directly.

New in 0.2.0, extended in 0.3.0

The request context on every storage call arrived in 0.2.0. 0.3.0 adds the optional parts (event log, leases, compare-and-set, message update, file index, deadlines, pinned-first), capabilities(), typed limit errors and the contract suite that holds all of them to one behaviour. A 0.2.x provider keeps working: everything new is optional and the server falls back where a part is missing (Upgrading to 0.3.0).

When to use it

• Your conversations already live in a database or service of your own.
• You must scope data by tenant or user inside the store (Request Context).
• You need a database none of the four built-in providers cover.

For PostgreSQL, MongoDB or SQLite, use the built-in providers.

The request context

Every data method takes the per-request RequestContext (see Request Context) as its last, optional argument. Use it to scope data by tenant/user. Providers that don't need it can leave the parameter out.

interface RequestContext {
  user?: { id: string; metadata?: Record<string, unknown> };
  data: Record<string, unknown>; // whatever resolveContext put there
}

Using BaseStorageProvider

import {
  BaseStorageProvider,
  type Conversation, type CreateConversationParams, type GetConversationsParams,
  type PaginatedResult, type RequestContext,
} from '@cognipeer/agent-server';

export class TenantStorage extends BaseStorageProvider {
  async connect() { /* open pool */ this._connected = true; }
  async disconnect() { this._connected = false; }

  // ids are generated by the base class
  protected async _createConversation(id: string, params: CreateConversationParams, ctx?: RequestContext): Promise<Conversation> {
    const tenantId = ctx?.data.tenantId as string;
    return db.conversations.insert({ id, tenantId, ...params });
  }

  protected async _getConversation(id: string, ctx?: RequestContext) {
    // never return another tenant's row, even if the id is known
    return db.conversations.findOne({ id, tenantId: ctx?.data.tenantId });
  }

  protected async _getConversations(params: GetConversationsParams, ctx?: RequestContext): Promise<PaginatedResult<Conversation>> {
    /* ... filter by tenant and params.userId ... */
  }

  // _updateConversation, _deleteConversation, _createMessage, _getMessages,
  // _getMessage, _deleteMessage, _saveFile, _getFile, _getFileContent,
  // _deleteFile, _createTask, _getTask, _getTasks, _updateTask, _deleteTask,
  // _createTaskResult, _getTaskResult
}

Hook signatures (all end with ctx?: RequestContext):

Group	Hooks
Conversations	_createConversation(id, params, ctx), _getConversation(id, ctx), _getConversations(params, ctx), _updateConversation(id, params, ctx), _deleteConversation(id, ctx)
Messages	_createMessage(id, params, ctx), _getMessages(conversationId, params, ctx), _getMessage(id, ctx), _deleteMessage(id, ctx)
Files	_saveFile(id, params, ctx), _getFile(id, ctx), _getFileContent(id, ctx), _deleteFile(id, ctx)
Tasks	_createTask(id, params, ctx), _getTask(id, ctx), _getTasks(params, ctx), _updateTask(id, params, ctx), _deleteTask(id, ctx)
Task results	_createTaskResult(id, params, ctx), _getTaskResult(taskId, ctx)
File index (optional)	_listFiles({ conversationId, limit, offset, order }, ctx), _purgeExpiredFiles({ before, limit }, ctx)
Compare-and-set (optional)	_updateConversationIf(id, ifVersion, params, ctx)
Message update (optional)	_updateMessage(id, { content?, metadata? }, ctx)
Events (optional)	_appendEvents(conversationId, events, ctx), _listEvents(conversationId, query, ctx), _eventSummaries(conversationIds, ctx), _pruneEvents(before, ctx)
Leases (optional)	_acquireLease(params, ctx), _renewLease(params, ctx), _releaseLease(params, ctx), _getLeases(keys, now, ctx), _markLeaseCancel(key, now, ctx), _listExpiredLeases(params, ctx)

What a provider has to do

The built-in providers all pass one contract suite (test/storage/contract.ts in the repository; copy it to test your own: runStorageContract(name, open) takes a function that opens a fresh store, and runs under vitest). Its sections are capabilities, conversations, messages, files, tasks, events, leases, compare-and-set updates, action deadlines, message updates, tasks that run in a conversation, and persistence. As shipped it expects every optional part (the built-in providers have them all), so trim the sections for a provider that has fewer. A provider that follows these rules behaves like the built-in ones to the server and to every client:

Area	Rule
Order	getConversations / getTasks: newest first unless order: 'asc'. getMessages: oldest first, order: 'desc' reverses it. Rows with the same timestamp keep their insertion order (keep a sequence next to the time).
Paging	PaginatedResult has total, limit, offset and hasMore = offset + data.length < total. getMessages is called with params possibly undefined; the default limit is 100.
Cursors	getMessages(id, { before }) / { after } (a message id): only messages strictly older / newer than that message, total counting that set. An id that is not a message of the conversation is a NotFoundError. The server pages GET /conversations/:id with them.
JSON fields	metadata, state, content, toolCalls, files, comments, usage: store and return what JSON.parse(JSON.stringify(x)) gives. No null for a left-out field, Dates as ISO strings, text that looks like JSON stays text.
Absent	userId, title, name, toolCallId, callbackUrl, error read as undefined when absent; files: [] as undefined; metadata / state default to {}; Task.comments is always an array of comments with Dates.
Copies	Do not hand out objects you keep (a provider that holds records in memory must copy them).
Messages	createMessage for a conversation that does not exist throws NotFoundError and stores nothing. A new message sets the conversation's updatedAt. deleteMessage keeps the message's files.
Files	getFileContent returns a Buffer. saveFile for an unknown conversation or message throws NotFoundError and stores no content; content and record are written together (or the content is taken back out). A file with a messageId belongs to that message's conversation. deleteConversation deletes the conversation's files and their content. The record keeps userId (who uploaded it), agentId and expiresAt when given, and reads them back as undefined when not.
File index	listFiles({ conversationId, limit?, offset?, order? }): the conversation's files, oldest first (order: 'desc' reverses it; files saved in the same millisecond keep their insertion order), as a PaginatedResult; an unknown conversation holds none. purgeExpiredFiles({ before, limit? }) deletes files whose expiresAt <= before - record and content, earliest expiry first, at most limit (default 500) - and resolves how many; files without expiresAt stay. Index (conversationId, createdAt) and expiresAt.
Tasks	createTaskResult replaces the task's existing result (one per task). updateTask / createTaskResult for an unknown task throw NotFoundError; an update that changes nothing is not an error. conversationId is stored, updatable and a getTasks filter.
Events	append is atomic and ordered: seq is 1, 2, 3 ... per conversation with no repeats and no gaps, also for appenders running at the same time, and keeps counting after a restart or a prune. (A provider that reserves numbers before it stores the events, as MongoDB does, may leave a hole after a failed append; it must then never hand a number out twice, and summary must not report a number whose event list cannot return.) An unknown conversation is a NotFoundError. list({ after, before, types, limit, order }) returns { events, hasMore, lastSeq } (lastSeq = the last event returned, or after for an empty page). summary gives lastSeq (the newest stored event: never ahead of what list returns) and stopReason per conversation. The log is deleted with the conversation.
Leases	acquireLease is atomic: free, expired or already yours is acquired (an expired one is returned as takenOver), otherwise { acquired: false, lease } with the holder. renewLease / releaseLease only work for the owner, renewLease not once expired. markLeaseCancel flags a live lease, and the owner sees cancelRequested when it renews. getLeases returns the live ones, listExpiredLeases the expired, oldest first, by key prefix.
Version	Conversation.version is 0 when created and goes up by one with every updateConversation (not with a new message). updateConversationIf(id, ifVersion, params) does the same update only if the stored version is ifVersion, in one atomic step, and resolves null otherwise (an unknown conversation is a NotFoundError). Of 20 writers that read the same version exactly one wins.
Deadlines	UpdateConversationParams.actionDueAt (a Date, or null to clear) is stored on the conversation and read back as Conversation.actionDueAt; getConversations({ actionDueBefore }) returns the conversations whose actionDueAt is at or before it, earliest first (ties in insertion order), with the usual paging. Index actionDueAt (only rows that have one).
Message update	updateMessage(id, { content?, metadata? }) replaces what it is given, moves the message's updatedAt and nothing else (not its place in the history, not the conversation); an unknown message is a NotFoundError.
Limits	Throw StorageLimitError (413) for a file or a state you cannot hold, ValidationError (400) for a value a column cannot hold, NotFoundError (404) for a missing parent. Not a driver error.
Capabilities	Implement capabilities() so the server can refuse a bad configuration at startup and skip a pause that would not fit.

BaseStorageProvider already applies the JSON, absent-value and limit rules to what flows through it: values are cloned through JSON on the way in, every record is normalised on the way out, a file or state over capabilities() is refused with a 413, and a file with a messageId is filed under that message's conversation. You implement the storage; the rest of the table is up to the _-hooks.

Capabilities

class MyStorage extends BaseStorageProvider {
  capabilities() {
    return {
      maxFileBytes: 5 * 1024 * 1024 * 1024, // largest single file you store
      maxStateBytes: 4 * 1024 * 1024,       // largest conversation state (UTF-8 bytes of its JSON)
      atomicUpdate: false,                   // compare-and-set conversation updates (implement `_updateConversationIf`, keep `version`)
      actionDeadlines: false,                // `getConversations({ actionDueBefore })` and `actionDueAt`: lets the sweep find lapsed approvals
      events: false,                         // a durable per-conversation event log (implement `_appendEvents` / `_listEvents`)
      leases: false,                         // per-conversation leases (implement the `_...Lease` hooks)
      fileIndex: false,                      // `listFiles` / `purgeExpiredFiles` (implement `_listFiles` / `_purgeExpiredFiles`)
      pinnedFirst: false,                    // `getConversations({ pinnedFirst: true })`: metadata.pinned === true first, across pages
    };
  }
}

createAgentServer throws when limits.maxFileSize or limits.maxStateBytes is above these, and does not pause a turn whose state would be bigger than maxStateBytes (see Storage Providers). A provider without capabilities() is treated as unbounded.

Events and leases (optional)

Both are optional. Without events the server keeps the last 2 000 events per conversation in memory (one node, lost on restart); without leases it serialises runs in-process only (one node). Implementing them makes Runs, Cancel & Reattach durable and multi-node:

class MyStorage extends BaseStorageProvider {
  capabilities() { return { ...super.capabilities(), events: true, leases: true }; }

  // hand out seq 1, 2, 3 ... atomically (a counter row per conversation is the usual way)
  protected async _appendEvents(conversationId: string, events: StoredEventInput[], ctx?: RequestContext): Promise<ConversationEvent[]> { /* ... */ }
  protected async _listEvents(conversationId: string, q: EventQuery, ctx?: RequestContext): Promise<ConversationEvent[]> { /* q.limit already includes one extra row for `hasMore` */ }
  protected async _eventSummaries(ids: string[], ctx?: RequestContext): Promise<Record<string, EventSummary>> { /* ... */ }
  protected async _pruneEvents(before: Date, ctx?: RequestContext): Promise<number> { /* ... */ }

  protected async _acquireLease(p: AcquireLeaseParams & { now: Date; expiresAt: Date }, ctx?: RequestContext): Promise<AcquireLeaseResult> { /* compare-and-set */ }
  // _renewLease, _releaseLease, _getLeases, _markLeaseCancel, _listExpiredLeases
}

BaseStorageProvider validates and normalises what flows through (event types and sizes, lease keys and lifetimes, the clock now / expiresAt), so the hooks only store. summary().stopReason is the stop_reason of the newest session.status_idle event, cleared by a later session.status_running: use nextStopReason(previous, events) (exported) so you agree with the built-in providers. If you implement StorageProvider directly, expose events (an EventStore) and the lease methods, and report capabilities().

The contract suite (runStorageContract) checks events and leases too.

Implementing the interface directly

When the data already lives elsewhere (another service, an existing conversation store) implement StorageProvider and map the calls:

import type { StorageProvider } from '@cognipeer/agent-server';

class ExistingStoreAdapter implements StorageProvider {
  constructor(private readonly store: MyStore) {}
  async connect() {}
  async disconnect() {}
  isConnected() { return true; }

  async getConversation(id: string, ctx?: RequestContext) {
    const row = await this.store.find(ctx!.data.tenantId as string, id);
    return row ? toConversation(row) : null;
  }
  // ...
}

Compare-and-set, deadlines and message updates (optional)

These make approvals safe when several things write one conversation at once - two answers to one pause, a sweep and an answer, a budget change and a turn:

• updateConversationIf(id, ifVersion, params) + Conversation.version (capabilities().atomicUpdate). The server reads the conversation, decides what to write, and writes it only if nobody changed it in between; otherwise it reads again and redoes the change. Without it the server writes plainly (fine for one node and one writer at a time).

• actionDueAt / getConversations({ actionDueBefore }) (capabilities().actionDeadlines). The server stores the earliest expiresAt of the open actions here, and the expiry sweep lists the due ones. Without it lapsed requests are still refused with 410, but only server.expirePendingActions({ conversationId }) handles them.

• updateMessage(id, { content?, metadata? }). The server records how an action ended on the paused assistant message (metadata.pendingActions) and stores the rating of an answer (metadata.feedback, see Chat experience). Without it the history keeps showing the request as it was asked, and a rating only reaches the feedback hooks.

• getConversations({ pinnedFirst: true }) (capabilities().pinnedFirst). A person can pin a conversation (metadata.pinned: true, with pinnedAt; these and titleGenerated are the server's keys: report what you stored in metadata, and the answer to PATCH /conversations/:id is laid over what updateConversation returns, so a stale echo does not show); with the flag the provider returns those first and the rest after them, each group in the requested order, so the order holds across pages. Without it the server only moves the pinned conversations of the page it got to the top.

• deleteMessage(id) is how a regeneration or an edit removes messages from the transcript (newest first; a NotFoundError for a message that is already gone is fine, which is what a host whose transcript lives elsewhere answers after its agent cut it back). getMessages is read with after / before (strictly newer / older than a message) to find what follows a message and where an answer sits.

BaseStorageProvider validates the arguments and normalises the result; implement the _ hooks and bump version in _updateConversation as well.

The file index (optional)

listFiles and purgeExpiredFiles serve files: the per-conversation file limit, GET /conversations/:id/files, files.attachNewUploads and the retention sweep. Without them (capabilities().fileIndex false) files are still stored, owned and served, but those four do nothing - the limit is not enforced, the listing shows only the files that messages carry, and expired files are not deleted. Implement _listFiles and _purgeExpiredFiles and report fileIndex: true.

Notes for scoped stores:

• Returning null from getConversation yields a 404 - use that for "not yours" instead of leaking existence. (Ownership is additionally enforced by the server through conversation.userId when resolveContext is configured.)
• The server checks a file's userId and its conversation before it shows a file (a file of someone else is a 404), so file ids are safe to hand out; scope tasks by ctx too if clients must not reach each other's ids (the server checks task.userId).
• getMessages is called with params possibly undefined, and with before / after when a client pages GET /conversations/:id.
• The server reads a whole conversation for the agent in pages of limits.maxPageSize (200): offset, limit and hasMore / total must be right.
• Methods must be safe to call concurrently; the server issues several per request.

Troubleshooting

Symptom	Cause and fix
500 "Storage provider is not connected"	A BaseStorageProvider subclass has to set this._connected = true in connect() and false in disconnect().
404 for a conversation that exists	_getConversation returned null for the context: a scoped store answers null for "not yours". Check ctx?.data is what you scope on, and let the sweep's context ({ data: { system: true } }) through.
Reattach and the event stream work, but are lost after a restart	The provider has no events (the server keeps 2 000 events per conversation in memory). Implement _appendEvents, _listEvents, _eventSummaries, _pruneEvents and report events: true.
409 conversation_busy is not enforced across instances	The provider has no leases; implement the _...Lease hooks and report leases: true.
Two answers to one approval overwrite each other	No compare-and-set: implement _updateConversationIf, keep version, and report atomicUpdate: true.
Lapsed approvals are refused with 410 but never swept	No actionDeadlines: store actionDueAt and support getConversations({ actionDueBefore }).
Per-conversation file limits and retention do nothing	No file index: implement _listFiles and _purgeExpiredFiles, report fileIndex: true.
createAgentServer throws about a limit	capabilities() reports a smaller maxFileBytes / maxStateBytes than limits.* or files.maxFileSize ask for.

Next Steps

• Storage Providers - what the built-in providers hold and how they differ
• Storage Providers API - every method of the interface
• Runs, Cancel & Reattach - what events and leases make possible