Mastra

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

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.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 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 uses S2 for topics beginning with agent.. This covers per-run topics such as agent.stream.<runId> 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.

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, but S2 is the authoritative state for durable topics:

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.
subscribe opens a read session at the live tail. If that session reconnects, it resumes from the next exact sequence number.
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.
getHistory reads retained events from an offset and restores each record’s seqNum as its event index.
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:

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 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 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:

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

Delivery semantics

Distributed leases

Cleanup and retention

Options

S2 Lite

Example

​Install

​Configure Mastra

​Reconnect to a run

​How persistence works

​Delivery semantics

​Distributed leases

​Cleanup and retention

​Options

​S2 Lite

​Example

Install

Configure Mastra

Reconnect to a run

How persistence works

Delivery semantics

Distributed leases

Cleanup and retention

Options

S2 Lite

Example