> ## 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 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 |
## 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.46 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](/docs/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` uses S2 for topics beginning with `agent.`. This covers per-run
topics such as `agent.stream.` and the per-thread topics Mastra uses for
coordination. Topics outside that prefix, and events published with
`localOnly`, use the inner local 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`, but S2 is the authoritative state
for durable topics:
1. `publish` serializes the event and appends it once to the topic's S2 stream.
Read sessions deliver the stored record to subscribers in the publishing
process and in other processes, restoring its `seqNum` as the event `index`.
2. `subscribe` opens a read session at the live tail. If that session
reconnects, it resumes from the next exact sequence number.
3. `subscribeFromOffset` opens a read session at the requested offset. The same
session replays retained records and then follows new records, avoiding a
replay-to-live handoff gap.
4. `getHistory` reads retained events from an offset and restores each record's
`seqNum` as its event `index`.
5. `clearTopic` stops this instance's subscriptions, clears local state, and
requests deletion of the corresponding S2 stream. Deletion failures are
logged rather than thrown.
The adapter does not keep event history, a replay cache, or a durable cursor in
process memory. Call `await pubsub.close()` during graceful shutdown to cancel
active read sessions.
## Delivery semantics
Publishing rejects if S2 does not acknowledge the append; durable topics do not
fall back to a process-local event. The S2 SDK's default append retry policy is
at-least-once, so an ambiguous timeout can produce a duplicate record. If that
is not acceptable, create the S2 client with `appendRetryPolicy:
'noSideEffects'` and pass the client to `S2PubSub`:
```ts theme={null}
import { S2 } from '@s2-dev/streamstore';
import { S2PubSub } from '@s2-dev/mastra-pubsub';
const client = new S2({
accessToken: process.env.S2_ACCESS_TOKEN!,
retry: { appendRetryPolicy: 'noSideEffects' },
});
const pubsub = new S2PubSub({
client,
basin: process.env.S2_BASIN!,
});
```
S2-delivered callbacks run one at a time per subscription. `S2PubSub` awaits
each callback before reading the next record, which preserves stream order and
applies backpressure when a callback is slow. A callback that never settles
stalls that subscription. If a callback rejects, the adapter logs the error and
considers the record delivered; broadcast observers do not provide
acknowledgement or callback-level redelivery.
Events use JavaScript's `JSON.stringify` semantics. Before appending, the
adapter validates the serialized record, so unsupported values such as
`data: undefined`, functions, symbols, bigints, or circular data reject
`publish` instead of leaving an unreadable record in the stream. Optional
nested properties whose value is `undefined` are omitted.
## Distributed leases
`getLeaseProvider()` returns the S2-backed lease provider used by
[Mastra's signals runtime](https://mastra.ai/docs/long-running-agents/signals#distributed-and-serverless-deployments)
to elect one owner per thread key across processes. The lease owner, expiry,
and nonce are encoded in the thread stream's S2 fencing token, so lease
operations remain atomic without an in-process cursor or a separate lease
stream.
S2 fence commands are filtered out of event history and live delivery, but
they still consume sequence numbers. Resume from `last.index + 1`, not from the
number of events received.
Do not trim a thread stream while its lease is held. `clearTopic` also deletes
the stream, so do not call it for an active thread topic.
## Cleanup and retention
When Mastra cleans up a run, it calls `clearTopic`, which deletes that run's S2
stream. Mastra's default cleanup window is 30 seconds; set `cleanupTimeoutMs`
to `0` if the stream should remain until explicit cleanup or S2 retention
removes it.
If a process exits before cleanup, its stream can remain. You can apply an
age-based [retention policy](/docs/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. A request whose offset has already been
trimmed fails instead of silently returning partial history.
## 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` | Local transport for non-S2 topics and explicit `localOnly` events. |
| `streamPrefix` | `mastra/durable/` | Prefix added to S2 stream names. |
| `topicPrefix` | `agent.` | Only topics with this prefix use S2, including per-run and per-thread topics. |
| `logger` | `console.error` | Logger used for swallowed or background pubsub failures. |
## S2 Lite
Install the S2 SDK for its environment helper:
```bash theme={null}
npm install @s2-dev/streamstore
```
To use [S2 Lite](/docs/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).