Mastra

S2 integrates with Mastra through @s2-dev/mastra-pubsub. The package provides an S2PubSub implementation for Mastra 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

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 or with the CLI:

s2 create-basin my-basin --create-stream-on-append --create-stream-on-read

Set the basin, access token, and model provider key:

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:

src/mastra/index.ts

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!,
  }),
});

src/mastra/agents/research-agent.ts

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.

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:

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:

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:

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.
getHistory reads the S2 stream from the requested offset and restores each record’s seqNum as its event index.
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 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:

npm install @s2-dev/streamstore

To use S2 Lite, set S2_ACCOUNT_ENDPOINT and S2_BASIN_ENDPOINT, then pass the parsed endpoints to S2PubSub:

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.

AI SDKs

Observability

Pipelines

Sync

Install

Configure Mastra

Reconnect to a run

How persistence works

Cleanup and retention

Options

S2 Lite

Example

​Install

​Configure Mastra

​Reconnect to a run

​How persistence works

​Cleanup and retention

​Options

​S2 Lite

​Example

Install

Configure Mastra

Reconnect to a run

How persistence works

Cleanup and retention

Options

S2 Lite

Example