Durable agents

Added in: @mastra/core@1.45.0

beta

Breaking changes may occur without a major version bump until the API is stable.

A durable agent wraps a regular Agent so the agentic loop runs inside a workflow. Events flow through PubSub, which means a client can disconnect and reconnect without missing chunks. The run state is persisted, so it survives process restarts.

When to use durable agents

Use a durable agent when any of the following apply:

• The client may drop and reconnect mid-stream (mobile, spotty networks, long-running calls).
• The agentic loop may outlive a single HTTP request (background research, multi-step tool use).
• You need an observe/reconnect API where a second client picks up a stream that a first client started.
• You want Inngest-powered execution with step memoization, retries, and monitoring.

For short-lived, request-scoped calls where the client stays connected, a regular Agent with stream() or generate() is simpler.

Quickstart

Wrap an existing agent with createDurableAgent() from @mastra/core/agent/durable:

src/mastra/agents/researcher.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'researcher',
  name: 'Researcher',
  instructions: 'You research topics thoroughly.',
  model: 'openai/gpt-5.6-sol',
})

export const durableResearcher = createDurableAgent({ agent })

Register the durable agent with Mastra and call stream():

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { durableResearcher } from './agents/researcher'

const mastra = new Mastra({
  agents: { durableResearcher },
})

const { output, runId, cleanup } = await durableResearcher.stream(
  'Research quantum computing advances in 2025',
)

for await (const chunk of output.fullStream) {
  // Process each chunk as it arrives
}

// Release PubSub subscriptions and clear the run from the registry.
// If you skip this, an automatic cleanup timer fires after the stream ends.
cleanup()

The returned runId identifies the execution. Pass it to observe() to reconnect from a different client. Visit the DurableAgent reference for the full configuration and method API.

How it works

A durable agent adds three layers on top of a regular agent:

1. Workflow execution: stream() serializes the messages and options into a workflow input, then triggers the agentic loop inside a durable workflow. The workflow runs the same loop as Agent.stream() but each step can be memoized and replayed.

2. PubSub streaming: The loop publishes chunks to a PubSub topic keyed by the run ID. The caller subscribes to that topic and pipes the chunks into a ReadableStream, while the cache replays any chunks missed during a disconnect.

3. Cache layer: An optional cache (in-memory by default, Redis or another backend in production) stores published events so that a late subscriber can catch up.

Execution variants

Mastra provides three factory functions that produce durable agents. They differ in how the workflow is executed:

Factory	Package	Best for
createDurableAgent()	@mastra/core	Local development and single-process servers. You get a stream you can await directly.
createEventedAgent()	@mastra/core	Background execution. The workflow starts without blocking, and you consume chunks through PubSub.
createInngestAgent()	@mastra/inngest	Production deployments. Inngest adds step memoization, retries, and a monitoring dashboard.

All three return an object you register with Mastra the same way as a regular agent. createDurableAgent() and createEventedAgent() return class instances that extend Agent. createInngestAgent() returns a Proxy-backed object that forwards Agent methods to the underlying agent. When a signal wakes an idle thread, all three start that run with the durable stream(). sendSignal() and sendNotificationSignal() both work this way.

In-process with createDurableAgent()

Wrap your agent and call stream(). You get a DurableAgentStreamResult back in the same process. No external infrastructure is required, so this is the fastest way to get started:

src/mastra/agents/durable.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'helper',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5.6-sol',
})

export const durableHelper = createDurableAgent({ agent })

Fire-and-forget with createEventedAgent()

The workflow starts in the background without blocking the caller. You still receive chunks through PubSub, so stream() returns a result you can consume. The HTTP handler that triggered the run doesn't need to wait for the workflow to finish:

src/mastra/agents/evented.ts

import { Agent } from '@mastra/core/agent'
import { createEventedAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'writer',
  instructions: 'You write articles.',
  model: 'openai/gpt-5.6-sol',
})

export const eventedWriter = createEventedAgent({ agent })

Stored agents created through the API

Agents created with createStoredAgent() opt in with a durable field on the agent config. The server wraps the agent with createDurableAgent() when it hydrates it, so no code deployment is needed:

await mastraClient.createStoredAgent({
  id: 'helper',
  name: 'Helper',
  instructions: 'You are a helpful assistant.',
  model: { provider: 'openai', name: 'gpt-5' },
  durable: true,
})

durable also accepts { maxSteps, cleanupTimeoutMs }. Only serializable options are accepted here, so snapshot persistence for API-created agents follows the server's recovery.durableAgents setting. Cache and pubsub are inherited from the server's Mastra instance, so configure distributed backends there if you need durability across replicas. Automatic recovery is still configured in code through recovery.durableAgents.

Inngest-powered with createInngestAgent()

Run the workflow on the Inngest platform. Each tool call becomes a memoized step that Inngest can retry independently, and you get a dashboard for monitoring runs:

src/mastra/agents/inngest.ts

import { Agent } from '@mastra/core/agent'
import { createInngestAgent } from '@mastra/inngest'
import { Inngest } from 'inngest'

const inngest = new Inngest({ id: 'my-app' })

const agent = new Agent({
  id: 'analyst',
  instructions: 'You analyze data.',
  model: 'openai/gpt-5.6-sol',
})

export const inngestAnalyst = createInngestAgent({ agent, inngest })

Visit the createInngestAgent() reference for the full API, including Inngest-specific options like PubSub and cache configuration.

Resumable streams

Durable agents support resumable streams through PubSub and an event cache. When a client disconnects mid-stream, the cache continues storing events. The same client can reconnect by calling observe() with the runId:

src/server/reconnect.ts

const { output } = await durableResearcher.observe(runId)

for await (const chunk of output.fullStream) {
  // Chunks from the run, including any missed while disconnected
}

Calling observe(runId) without an offset is the supported general reconnect flow. It replays all available cached events before delivering live events. The optional offset is an inclusive, zero-based PubSub event index, not a chunk number, and the public stream doesn't expose it. If an offset skips earlier text deltas, aggregated text is partial and structured output may be invalid. On transports with numeric offset support, an offset beyond the highest retained index also skips live events with lower indexes. Events arrive only when the published index reaches the requested offset. If the index counter restarts below that offset and never reaches it, observation remains pending indefinitely. See observe(runId, options?) for the full offset and transport behavior.

Observing doesn't own the run. When the loop ends, or you leave it early with break, return, or an error, the observer unsubscribes and the run keeps going for anyone else watching it. To stop observing from outside the loop, for example when the client disconnects, call detach():

src/server/reconnect.ts

const { output, detach } = await durableResearcher.observe(runId)

request.signal.addEventListener('abort', detach, { once: true })

if (request.signal.aborted) detach()

for await (const chunk of output.fullStream) {
  // Stops when the run finishes or the client disconnects
}

If you pass idleTimeoutMs and no chunks arrive for that long, Mastra checks isAlive when you provide it. If isAlive returns true or throws, the timer restarts and the stream stays open. Otherwise the stream ends with an error chunk and detaches the observer. The run itself is only cleaned up when isAlive returns false.

createDurableAgent() and createEventedAgent() use an in-memory cache by default, which means resumable streams work within a single process. For production, provide a persistent cache backend (e.g., Redis) so cached events survive process restarts:

src/mastra/agents/durable-with-cache.ts

import { createDurableAgent } from '@mastra/core/agent/durable'
import { RedisServerCache } from '@mastra/redis'
import Redis from 'ioredis'

const cache = new RedisServerCache({ client: new Redis('redis://localhost:6379') })

export const durableAgent = createDurableAgent({
  agent,
  cache,
})

createInngestAgent() doesn't enable caching by default. Pass a cache option or register the agent with a Mastra instance that has a serverCache configured to enable resumable streams.

For cached topics, each published event is recorded in the cache before it's delivered live, so publish latency is bounded by the round-trip to the cache. If the cache write fails, the event is still delivered live but can't be replayed. @mastra/redis and @mastra/valkey record each event in a single round-trip using a Lua script. @mastra/redis falls back to separate commands when the client has no evalScript or when Redis Cluster rejects the multi-key script. Per-run workflow watch events (workflow.events.v2.*) are never cached. If the cache is remote (for example, in another region) and you don't need to resume a topic, use shouldCache to publish that topic straight through:

src/mastra/agents/durable-hot-topics.ts

export const durableAgent = createDurableAgent({
  agent,
  cache,
  // Skip the replay cache for the per-chunk stream topic; other topics stay resumable.
  shouldCache: topic => !topic.startsWith('agent.stream.'),
})

Streaming with background tasks

Durable agents support the same untilIdle option as regular agents. When untilIdle is set, stream() keeps the connection open across background-task continuations until the agent is idle:

src/mastra/run.ts

const { output, cleanup } = await durableAgent.stream('Research and summarize the topic', {
  untilIdle: true,
  memory: { thread: 'thread-1', resource: 'user-1' },
})

for await (const chunk of output.fullStream) {
  // Chunks from the initial turn AND any follow-up turns triggered by
  // background task completions
}

cleanup()

Pass { maxIdleMs } to customize the idle timeout (defaults to 5 minutes):

await durableAgent.stream('Research topic', {
  untilIdle: { maxIdleMs: 30_000 },
  memory: { thread: 'thread-1', resource: 'user-1' },
})

Visit Background tasks for the full background task guide, including configuration, subagents, and suspend/resume.

Cleanup

Every stream() and observe() call returns a cleanup function that tears the run down: it unsubscribes from PubSub, removes the run from the internal registry, and deletes the run's cached events so nobody can replay it afterward.

When you don't call cleanup() yourself, Mastra calls it cleanupTimeoutMs after the run finishes, errors, or is aborted. Detaching an observer or reaching a plain idleTimeoutMs doesn't start this timer. Set cleanupTimeoutMs: 0 to turn the timer off.

Call cleanup() from the process that started the run. To stop watching a run from observe(), use detach() or leave the loop instead.

Tool approval

Durable agents support tool approval (human-in-the-loop). When a tool call requires approval, the workflow suspends, emits an onSuspended callback, and waits for the caller to resume with resume():

src/mastra/run.ts

const { output, runId, cleanup } = await durableAgent.stream('Delete the old records', {
  requireToolApproval: true,
  onSuspended: ({ toolCallId, toolName, args }) => {
    // Notify the user and ask for approval
  },
})

Resume the suspended run after approval:

await durableAgent.resume(runId, { approved: true })

Crash recovery

Durable agents can checkpoint each run's state to storage while it executes. If the server process crashes mid-run, the run remains in running status in storage with no automatic retry, and on the next server start you can re-drive these orphaned runs so they pick up where they left off. For orderly shutdowns such as rolling deploys, the generated server can also drain in-flight turns before exiting. See graceful shutdown and rolling deploys.

Under the default persistence policy, these running checkpoints are written when crash recovery is enabled. With recovery.durableAgents: 'off' (the default), durable agents persist snapshots for pending, paused, and suspended runs (the artifacts human-in-the-loop resume depends on) and skip the per-step running writes until mastra.shutdown() starts. A custom shouldPersistSnapshot predicate replaces the default policy: one that includes running writes those checkpoints at every step, and one that excludes running writes none, including during shutdown. See snapshot persistence for the full policy and how to override it.

Durable agent runs are excluded from the generic boot-time restart of active workflow runs. The only automatic recovery path for durable agent runs is recovery.durableAgents: 'auto', which claims each orphaned run and registers its thread runtimes before re-driving it. See run ownership.

warning

Automatic and manual recovery re-run the agentic loop from the last persisted snapshot, which can reissue LLM calls (with real cost) and re-execute tool calls. A sub-agent delegation is one generated agent-<name> tool call, and recovery doesn't checkpoint the LLM or tool calls inside that delegation independently. If recovery replays the delegation step, the entire sub-agent run starts again and can repeat inner LLM calls and already-completed tool side effects. Make tools, including tools used by sub-agents, idempotent, or deduplicate their external side effects.

Snapshot persistence

The shouldPersistSnapshot option on createDurableAgent() (also accepted in the agent-level durable config) controls which workflow snapshots a durable agent writes. The default policy:

• Always persists pending, paused, and suspended snapshots. These are what resume() and tool approval read, so human-in-the-loop flows work without configuration.
• Persists running checkpoints at every step when recovery.durableAgents is 'auto'. These per-step writes exist solely to make in-flight runs recoverable after a crash, so with recovery off they're skipped until shutdown.
• Persists a running checkpoint at each step boundary a run reaches after mastra.shutdown() starts, even with recovery off. A run that reaches a step boundary before the drain window ends can then be resumed later with recover() or recoverActiveRuns(). A run blocked in a single LLM or tool call for the whole drain gets no shutdown checkpoint.

When the agent is configured with a fixed memory instance (memory: new Memory(...)), a running checkpoint doesn't repeat messages memory already stores. Messages recalled from memory and messages the run saved itself (for example with savePerStep: true) are stored as small references to their memory rows instead of their full contents, as long as they still match the stored row (a message changed after saving stays inline). This applies to every copy of the conversation a checkpoint holds, including the run input and step results. A long run's checkpoints grow much more slowly as a result. Agents that resolve memory with a function keep the full transcript in each checkpoint, because the memory a resumed run resolves may differ.

To keep crash-recovery checkpoints without enabling automatic recovery (for example, when you call recoverActiveRuns() yourself behind a leader election), pass a predicate that includes running:

src/mastra/agents/durable.ts

export const durableResearcher = createDurableAgent({
  agent,
  shouldPersistSnapshot: ({ workflowStatus }) =>
    ['pending', 'paused', 'suspended', 'running'].includes(workflowStatus),
})

Mastra logs a warning when a custom predicate excludes suspended or paused, which breaks human-in-the-loop resume, or excludes running while recovery.durableAgents is 'auto', which makes the agent invisible to automatic recovery.

Evented agents always persist the full snapshot set: the evented engine coordinates workers through storage, so the running row is part of its execution model. Inngest agents always persist only suspended snapshots, because Inngest's own replay provides durability. Both accept shouldPersistSnapshot for API symmetry but log a warning and ignore it.

Automatic recovery

Set recovery.durableAgents to 'auto' in the Mastra config. The deployer calls recoverAllDurableAgents() on boot, alongside the workflow run restarts (the two run concurrently):

src/mastra/index.ts

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  storage: new PostgresStore({ connectionString: process.env.DATABASE_URL! }),
  recovery: { durableAgents: 'auto' },
})

On startup, this discovers every registered durable agent with runs stuck in running status and re-drives them from the last persisted snapshot.

Recovery doesn't take over a run that another execution still drives. It skips the run and checks it again once that execution's claim could have lapsed. A run whose process crashed shortly before the restart is still recovered, without another call.

Manual recovery

If you need finer control, such as running recovery on a schedule, call the methods directly:

// Recover all durable agents
const result = await mastra.recoverAllDurableAgents()
console.log(
  `Recovered ${result.recovered} runs (${result.succeeded} ok, ${result.failed} failed, ${result.skipped} still live)`,
)

// Recover a specific agent
const { recovered } = await durableAgent.recoverActiveRuns()
for (const run of recovered) {
  if (run.status === 'skipped') console.log(`${run.runId} is still live: ${run.reason}`)
}

// Recover a single known run
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })

recoverAllDurableAgents() returns counts and re-checks skipped runs itself. recoverActiveRuns() returns each run's outcome as success, failed, or skipped, and doesn't re-check skipped runs. A run skipped because another process drives it carries retryAt, the time after which its claim lapses unless renewed.

Manual recovery reads the same running checkpoints as automatic recovery. For a DurableAgent using the default snapshot policy with recovery.durableAgents set to 'off', in-flight runs are only checkpointed at step boundaries reached after mastra.shutdown() starts. A run that crashes before then, or doesn't reach a step boundary during the drain, has no checkpoint, so listActiveRuns() and recoverActiveRuns() don't find it. See snapshot persistence.

To take over a run that's still live, such as one held by a hung execution, call recover() with force: true:

const { output } = await durableAgent.recover('run-abc-123', { force: true })

The execution that held the run loses it and stops, and streams for the run drop the events it still publishes. Without force, recover() throws DURABLE_AGENT_RUN_ACTIVE for a live run. Even with force, recover() refuses a run that the same process is executing.

Run ownership

Every stream(), generate(), resume(), and recover() call claims the run before it executes and renews the claim every 10 seconds. A claim lapses 30 seconds after its last renewal, so a run whose process crashed becomes recoverable within 30 seconds. Until then, recovery treats the run as live, and a stream() or resume() call for that run waits up to 5 seconds for the claim, then throws DURABLE_AGENT_EXECUTION_CONFLICT. mastra.shutdown() stops the runs still executing after its drain window without further writes and releases their claims. The next boot then recovers them without waiting for the claims to lapse.

Enforcement of the claim depends on the workflows store. When the store supports run fencing, the claim is a record in that store, and each write is checked against the claim in the same operation. Once another execution claims the run, writes from the execution that lost it are rejected. That execution's message, thread, and working memory writes are rejected only if the memory store supports run fencing too, so they still land when the agent's Memory uses a storage adapter without it. These storage adapters support run fencing:

• InMemoryStore from @mastra/core/storage
• @mastra/pg
• @mastra/libsql and @mastra/turso
• @mastra/mysql
• @mastra/mssql
• @mastra/oracledb
• @mastra/spanner
• @mastra/dsql
• @mastra/cloudflare-d1
• @mastra/cloudflare, with CloudflareDOStorage
• @mastra/convex
• @mastra/mongodb, on a replica set or sharded cluster
• @mastra/dynamodb
• @mastra/redis
• @mastra/upstash
• @mastra/valkey

On Convex, run fencing runs in the Mastra server functions you deploy. Redeploy your Convex functions after upgrading @mastra/convex. Until you do, durable agent runs fail with an error asking you to redeploy, while other writes keep working.

Stores that support run fencing keep one claim record per run, plus one memory fence record per run with memory. Stores with retention support prune them with the workflows.runOwnership and memory.runFences retention keys. Redis, Upstash, and Valkey expire them on their own instead: a claim 7 days after its last claim, renewal, or release, and a fence 7 days after it's retired. A run's first generation comes from the store's clock, so a run that reuses a runId after its records were pruned still outranks the fences and stream events of its earlier executions.

With any other workflows store, the claim is a lease on the PubSub backend. The lease keeps two executions from claiming the run at once, but writes from an execution that lost the run aren't rejected. A run taken over while its previous execution was still live can end with that execution's output. When an execution loses its lease and no other execution has taken the run, the run stays held for up to 15 more seconds. Until then, recover() without force throws DURABLE_AGENT_RUN_ACTIVE with retryAt set to when the hold is sure to have ended. Mastra logs a warning when it recovers runs against a workflows or memory store without run fencing.

A lease only coordinates the processes that share it. EventEmitterPubSub, the default, covers one process. UnixSocketPubSub covers the processes on one host. RedisStreamsPubSub and ValkeyStreamsPubSub cover every instance connected to the same server. GoogleCloudPubSub doesn't provide leases, so with a store that lacks run fencing every claim succeeds, and recovery can take over a run that's still live.

Multi-instance deployments

Every replica can start with recovery.durableAgents: 'auto'. Replicas that boot together race to recover the same orphaned runs, but only one claim succeeds per run and the other replicas skip it. This requires a claim that every replica shares: a workflows store that supports run fencing, or a PubSub backend whose lease every replica shares. Without either, gate recovery behind your own leader election or run it from a single replica.

Related

• DurableAgent reference
• createInngestAgent() reference
• Background tasks
• Inngest deployment guide
• Agent overview
• Worker overview