> ## Documentation Index > Fetch the complete documentation index at: https://s2.dev/docs/llms.txt > Use this file to discover all available pages before exploring further. # Mastra > Persist and replay Mastra durable-agent streams with S2. S2 integrates with [Mastra](https://mastra.ai) through [`@s2-dev/mastra-pubsub`](https://www.npmjs.com/package/@s2-dev/mastra-pubsub). The package provides an `S2PubSub` implementation for [Mastra durable agents](https://mastra.ai/blog/introducing-durable-agents). Each durable-agent topic maps to an S2 stream. When `S2PubSub` appends an event, the S2 sequence number becomes the event's Mastra `index`. Live delivery and replay therefore use the same index, which lets Mastra replay and deduplicate events without a separate counter. | Mastra operation | S2 operation | | --------------------- | ----------------------------------------- | | Publish an event | Append a record | | Assign an event index | Use the appended record's sequence number | | Replay from an index | Read from a sequence number | | Store a run's events | Use one stream for the run topic | A durable agent reconnecting after a browser refresh ## Install ```bash theme={null} npm install @s2-dev/mastra-pubsub @mastra/core ``` The package requires Node.js 22.13 or later and `@mastra/core` 1.45 or later in the 1.x release line. `@mastra/core` is a peer dependency. Create a basin with **Create Stream on Append** and **Create Stream on Read** enabled. These settings let `S2PubSub` create a run stream when it first writes or reads the topic. You can create the basin in the [dashboard](https://s2.dev/dashboard) or with the [CLI](/cli/basins#create-basin): ```bash theme={null} s2 create-basin my-basin --create-stream-on-append --create-stream-on-read ``` Set the basin, access token, and model provider key: ```bash theme={null} export S2_ACCESS_TOKEN="..." export S2_BASIN="my-basin" export OPENAI_API_KEY="..." ``` ## Configure Mastra Create an `S2PubSub` and pass it to the Mastra instance that registers your durable agent: ```ts src/mastra/index.ts theme={null} import { Mastra } from '@mastra/core'; import { S2PubSub } from '@s2-dev/mastra-pubsub'; import { durableResearchAgent } from './agents/research-agent'; export const mastra = new Mastra({ agents: { durableResearchAgent }, pubsub: new S2PubSub({ accessToken: process.env.S2_ACCESS_TOKEN!, basin: process.env.S2_BASIN!, }), }); ``` ```ts src/mastra/agents/research-agent.ts theme={null} import { Agent } from '@mastra/core/agent'; import { createDurableAgent } from '@mastra/core/agent/durable'; const agent = new Agent({ id: 'research', name: 'research', instructions: 'Answer research questions concisely.', model: 'openai/gpt-4o', }); export const durableResearchAgent = createDurableAgent({ agent }); ``` Registered durable agents use the Mastra instance's pubsub. By default, `S2PubSub` persists topics beginning with `agent.stream.`; other topics only use the live-delivery transport. ## Reconnect to a run Starting a durable-agent stream returns a `runId`. Keep that ID where a reconnecting client can retrieve it. ```ts theme={null} const started = await durableResearchAgent.stream( 'Summarize the latest on durable streams.', ); console.log(started.runId); for await (const chunk of started.output.fullStream) { // Send chunks to the client as they arrive. } ``` After a browser refresh or dropped connection, call `observe` with the same `runId`: ```ts theme={null} const observed = await durableResearchAgent.observe(runId); for await (const chunk of observed.output.fullStream) { // Replayed events arrive before new live events. } ``` Pass an `offset` when the caller already has part of the stream. The offset is the first event index to replay, so use `42` if index `41` was the last event processed: ```ts theme={null} const observed = await durableResearchAgent.observe(runId, { offset: 42 }); ``` ## How persistence works `S2PubSub` extends Mastra's `CachingPubSub` and replaces its in-memory history operations for persisted topics: 1. `publish` serializes the event and appends it to the topic's S2 stream. It adds the returned `seqNum` as the event `index`, then sends the event through the inner pubsub for live delivery. 2. `getHistory` reads the S2 stream from the requested offset and restores each record's `seqNum` as its event `index`. 3. `clearTopic` deletes the corresponding S2 stream. If an append fails, `S2PubSub` logs the error and still sends the event through the inner pubsub without an index. That event is available to live subscribers but cannot be replayed from S2. ## Cleanup and retention When Mastra cleans up a run, it calls `clearTopic`, which deletes that run's S2 stream. Do not treat these streams as permanent run history. If a process exits before cleanup, its stream can remain. You can apply an age-based [retention policy](/concepts/configs#stream-config) and `delete_on_empty` in the basin's default stream configuration to remove these orphaned streams. Keep the retention period longer than the maximum run time and reconnection window you support. ## Options `new S2PubSub(config, options?)` | config | description | | ------------- | ---------------------------------------------------------- | | `accessToken` | S2 access token. Provide this or `client`. | | `client` | Existing `S2` client. Takes precedence over `accessToken`. | | `basin` | Basin that stores the durable-agent streams. | | `endpoints` | Optional endpoint overrides, such as S2 Lite endpoints. | | option | default | description | | -------------- | -------------------- | ----------------------------------------------------------------------- | | `inner` | `EventEmitterPubSub` | Pubsub used for live delivery. S2 handles persisted history and replay. | | `streamPrefix` | `mastra/durable/` | Prefix added to S2 stream names. | | `topicPrefix` | `agent.stream.` | Only topics with this prefix are persisted to S2. | | `logger` | `console` | Logger used when persistence fails. | ## S2 Lite Install the S2 SDK for its environment helper: ```bash theme={null} npm install @s2-dev/streamstore ``` To use [S2 Lite](/s2-lite), set `S2_ACCOUNT_ENDPOINT` and `S2_BASIN_ENDPOINT`, then pass the parsed endpoints to `S2PubSub`: ```ts theme={null} import { S2PubSub } from '@s2-dev/mastra-pubsub'; import { S2Environment } from '@s2-dev/streamstore'; const { endpoints } = S2Environment.parse(); const pubsub = new S2PubSub({ accessToken: process.env.S2_ACCESS_TOKEN!, basin: process.env.S2_BASIN!, endpoints, }); ``` ## Example The package repository includes a runnable example adapted from Mastra's durable-agents example: [`examples/durable-agents`](https://github.com/s2-streamstore/mastra-pubsub/tree/main/examples/durable-agents).