DurableAgent
Breaking changes may occur without a major version bump until the API is stable.
DurableAgent wraps an existing Agent with durable execution and resumable streams. It runs the agentic loop so that a client can disconnect and reconnect without missing events, and it streams those events over PubSub. Use it when a run must outlive a single request or survive a dropped connection.
Create one with the createDurableAgent factory, or use createEventedAgent for fire-and-forget execution on the built-in workflow engine. For Inngest-powered execution, use createInngestAgent from @mastra/inngest.
Usage exampleDirect link to Usage example
import { Mastra } from '@mastra/core'
import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'
const agent = new Agent({
id: 'my-agent',
name: 'My Agent',
instructions: 'You are a helpful assistant',
model: 'openai/gpt-6.1-sol',
})
const durableAgent = createDurableAgent({ agent })
export const mastra = new Mastra({
agents: { myAgent: durableAgent },
})
Stream a response and read the result. The cleanup function unsubscribes from PubSub when you are done with the run:
const { output, runId, cleanup } = await durableAgent.stream('Hello!')
const text = await output.text
cleanup()
Plain Agent stream configurationDirect link to Plain Agent stream configuration
The durable option on a plain Agent controls resumable streams only. It never wraps the agent or enables checkpoints. Use createDurableAgent({ agent }) explicitly for durable execution, and pass execution options such as maxSteps and shouldPersistSnapshot to that factory.
createDurableAgent(options)Direct link to createdurableagentoptions
Wraps an Agent with durable execution and resumable streams. This is the recommended way to create a DurableAgent.
import { createDurableAgent } from '@mastra/core/agent/durable'
const durableAgent = createDurableAgent({ agent })
Returns: DurableAgent
ParametersDirect link to Parameters
agent:
id?:
name?:
cache?:
false to disable caching, which makes streams non-resumable.pubsub?:
mastra.pubsub. When omitted, a registered agent publishes on mastra.pubsub and an unregistered agent uses an in-process EventEmitterPubSub.maxSteps?:
shouldCache?:
false to publish a topic straight to the underlying PubSub without recording it; subscribers of that topic receive live events only and cannot resume from an offset. Useful for trading replay for minimum publish latency on hot topics when the cache is remote (for example, cross-region Redis). Run-local topics are always excluded, regardless of this option.shouldPersistSnapshot?:
pending, paused, and suspended (required for human-in-the-loop resume) and persists running checkpoints only when the Mastra instance is configured with recovery.durableAgents: 'auto'. Pass a predicate that includes running to keep crash-recovery checkpoints for manual listActiveRuns() / recoverActiveRuns() without enabling automatic recovery. Mastra logs a warning when a predicate excludes suspended or paused, or excludes running while recovery.durableAgents is 'auto'.createEventedAgent(options)Direct link to createeventedagentoptions
Wraps an Agent with fire-and-forget durable execution on the built-in evented workflow engine. Like createDurableAgent, it returns a result you stream from, but the run is started without being awaited: events on the Mastra instance's PubSub drive execution independently of the caller. Register the agent on a Mastra instance with storage to get evented execution; without one, the agent falls back to the default in-process engine and logs a warning. It doesn't accept id or name overrides.
import { createEventedAgent } from '@mastra/core/agent/durable'
const eventedAgent = createEventedAgent({ agent })
Returns: EventedAgent (a subclass of DurableAgent)
ParametersDirect link to Parameters
agent:
cache?:
false to disable caching.pubsub?:
mastra.pubsub instead, because evented steps publish there.maxSteps?:
shouldCache?:
false to publish a topic straight to the underlying PubSub without recording it; subscribers of that topic receive live events only and cannot resume from an offset. Useful for trading replay for minimum publish latency on hot topics when the cache is remote (for example, cross-region Redis). Run-local topics are always excluded, regardless of this option.shouldPersistSnapshot?:
createDurableAgent, but ignored: the evented engine requires the full snapshot set (pending, paused, suspended, running) because the initial running write creates the base row that suspend-merges and multi-worker coordination build on. A warning is logged if set.Constructor parametersDirect link to Constructor parameters
The DurableAgent class accepts the same options as createDurableAgent, plus cleanupTimeoutMs. Prefer the factory unless you need to subclass.
agent:
id?:
name?:
cache?:
false to disable caching.pubsub?:
mastra.pubsub. When omitted, a registered agent publishes on mastra.pubsub and an unregistered agent uses an in-process EventEmitterPubSub.maxSteps?:
shouldCache?:
false to publish a topic straight to the underlying PubSub without recording it; subscribers of that topic receive live events only and cannot resume from an offset. Useful for trading replay for minimum publish latency on hot topics when the cache is remote (for example, cross-region Redis). Run-local topics are always excluded, regardless of this option.shouldPersistSnapshot?:
pending, paused, and suspended (required for human-in-the-loop resume) and persists running checkpoints only when the Mastra instance is configured with recovery.durableAgents: 'auto'.cleanupTimeoutMs?:
cleanup() call. Auto-cleanup does not fire on suspended events.MethodsDirect link to Methods
ExecutionDirect link to Execution
stream(messages, options?)Direct link to streammessages-options
Streams a response using durable execution. Returns immediately with a result whose output produces events as the run progresses.
const { output, runId, cleanup } = await durableAgent.stream('Hello!', {
onChunk: chunk => console.log(chunk),
onFinish: result => console.log('done', result),
})
const text = await output.text
cleanup()
Returns: Promise<DurableAgentStreamResult>
resume(runId, resumeData, options?)Direct link to resumerunid-resumedata-options
Resumes a suspended run, for example after a tool approval. Pass the runId from the original stream and the data the run was waiting on. Throws if no registry entry exists for the run. If another execution holds the run, resume() waits up to 5 seconds for it to finish, then throws DURABLE_AGENT_EXECUTION_CONFLICT. See run ownership.
const { output, cleanup } = await durableAgent.resume(runId, {
approved: true,
})
await output.text
cleanup()
Returns: Promise<DurableAgentStreamResult>
observe(runId, options?)Direct link to observerunid-options
Reconnects to an existing run. Use this after a network disconnection. For the supported general reconnect flow, omit offset to replay all available cached events before receiving live events.
The observing agent must have the same effective id as the agent that started the run, both for live observations in the same process and for persisted observations from another process. The effective ID is the DurableAgent's id, including any ID override, rather than the wrapped agent's original ID. Different DurableAgent instances with the same effective ID can observe the same run.
If the live or persisted run identifies a different owner agent ID, observe() throws DURABLE_AGENT_OBSERVE_AGENT_MISMATCH. Legacy live entries without owner metadata retain their existing behavior.
For runs configured with structuredOutput, an observer in another process reconstructs output.object from the persisted schema and replayed text.
const { output } = await durableAgent.observe(runId, {
onChunk: chunk => console.log(chunk),
})
await output.text
The offset option is an inclusive, zero-based PubSub event index. It counts every cached event published to the run topic, including lifecycle events, rather than counting chunks. A chunk read from the run log carries the index of its event as chunk.index, on stream(), observe(), and the HTTP /stream and /observe endpoints. To resume after a dropped connection, pass index + 1 of the last chunk that has an index, or omit offset if nothing arrived. Lifecycle events take an index but produce no chunk, so indexes have gaps and a chunk count isn't a valid offset.
Some chunks carry no index. The client builds object chunks itself rather than reading them from the log. An abort chunk is sent beside the finish of the same run-log entry, and only the finish carries the index, so a client that reconnects between the two gets the abort again.
A run that used a structuring model publishes its object-result to the run log, so that chunk carries an index and replays like any other. A core observe() caller holding the schema can also rebuild structured output from replayed text, but gets object-result only when it replays the text that produced it. HTTP /observe replays the log and doesn't rebuild it. To rebuild it after a reconnect, observe without an offset and skip chunks whose index you already processed. That replay also rebuilds object chunks you already received, and they have no index to skip by. Each object chunk is a snapshot of the object so far, so keep the latest one. Indexes come from the replay cache, so a chunk whose cache write failed, or any chunk when caching is off, has no index, so no offset can resume at it. Omitting offset replays whatever the transport itself retains, such as a Redis stream, and can repeat chunks you already received. The paragraphs below cover offsets past the end and transports without numeric offsets.
index is additive and offset keeps its meaning, so code that already passes a log position, or replays from 0, needs no change. Code that passed a count of chunks received should switch to the last index + 1.
The following example records the resume offset while streaming and passes it to observe() after a reconnect:
let offset: number | undefined
const { output } = await durableAgent.stream(messages, { runId })
for await (const chunk of output.fullStream) {
if (chunk.index === undefined) continue
offset = chunk.index + 1
}
// After a dropped connection:
const { output: resumed } = await durableAgent.observe(runId, { offset })
A non-zero offset can skip earlier events. Aggregated values are built only from chunks delivered to that observer. If the offset skips earlier text deltas, output.text is partial and structured output may be incomplete or fail validation or parsing.
When the PubSub transport supports numeric offsets, the observer receives numbered events only when the published index reaches the requested offset. An offset beyond the highest retained index replays nothing and also skips live events with lower indexes.
An event without an index, such as one whose cache write failed, is delivered as it arrives whatever the offset.
If the retained log expires or is cleared and numbering restarts at 0 while an observer is attached, that observer resyncs to the new log and receives its events, including ones below the requested offset. Detection is best effort. If the cache can't be read at that moment, or the new log has numbered past the observer's position before its first event arrives, the observer can miss the restart and skip those events instead. An observer that reconnects after the restart with an offset from the old log can't tell the logs apart and skips the new log's events below that offset, so omit offset to replay the new log.
On a transport without numeric offset support, such as RedisStreamsPubSub with caching off, observe() with any offset, including 0, starts with new live events and doesn't replay retained events. This is the same in core and over HTTP /observe. Omit offset to replay them. (HTTP /observe on a plain Agent replays the retained events instead, which can repeat events already received.)
Indexed replay is limited to events still available in the configured cache. The cache has nothing to replay when caching is disabled, shouldCache excludes the run topic, or retained events have been cleared. Omitting offset then replays only what the transport itself retains, if anything.
A failed cache write loses only that event from the cache. It has no index, and the events written before and after it still replay when the cache numbers and appends an event in one step. The in-memory and Valkey caches always do. RedisServerCache does while Redis scripting works, and switches to two steps when the client has no eval or Redis Cluster rejects the script with CROSSSLOT. A custom cache that keeps the default listPushIndexed() also takes two steps. A two-step cache can leave a gap that makes offset replay skip later events, so replay such a log without an offset and skip indexes already processed. Cache replay can't recover an event the cache never stored. A transport that retains it, such as RedisStreamsPubSub, can still deliver it without an index.
An evented agent (createEventedAgent()) on a cross-process PubSub, such as Redis Streams, can use the default in-memory server cache when its steps run in the process that started the run. That process numbers the events. Use a shared cache such as RedisServerCache on the Mastra instance when steps run in another process, such as a dedicated worker started with workers: true. That process can't share the in-memory counter, so its events aren't numbered and an offset can't skip them. Mastra logs a warning in the worker when this happens. Also use a shared cache when an observer reconnects through another process, such as a different server replica, because it can't replay from the starting process's in-memory cache.
By default observe() waits indefinitely for events. If the process running the run stops unexpectedly, the run stops producing events but never emits a completion event, so the observed stream would wait forever. Pass idleTimeoutMs to end the stream after a bounded period of silence. Before timing out, observe() can consult an optional isAlive check. Return true while work continues, such as during a long-running tool call or a pause for human input. A false result ends the stream with an error, as does omitting isAlive. A transient throw from isAlive is treated as "still alive", so a momentary check failure never ends a live stream.
const { output } = await durableAgent.observe(runId, {
idleTimeoutMs: 30_000,
isAlive: () => runHeartbeat.isFresh(runId),
})
Ending a run on idle timeout runs the same cleanup as a run that errors (see the warning below), so its cached state is released rather than retained. Both options are opt-in. Omit them for the previous wait-indefinitely behavior.
Returns: Promise<DurableAgentStreamResult>
The cleanup() returned by observe() destroys the run's registry entries and cached events. Only call it when you are done with the run. If the run is suspended and you intend to resume later, don't call cleanup(). Let the auto-cleanup timer handle it after the run finishes or errors. Auto-cleanup doesn't fire on suspended events.
prepare(messages, options?)Direct link to preparemessages-options
Prepares a run for durable execution without starting it. Registers the run in the internal registry and returns the serialized workflow input. Use this when you need to control when and how the workflow is triggered.
const { runId, messageId, workflowInput, threadId, resourceId } = await durableAgent.prepare(
'Summarize the document',
{
memory: { threadId: 'thread-1', resourceId: 'user-1' },
},
)
Returns:
interface PrepareResult {
runId: string
messageId: string
workflowInput: any
registryEntry: object
threadId?: string
resourceId?: string
}
CancellationDirect link to Cancellation
abortRunStream(runId)Direct link to abortrunstreamrunid
Stops a run by ID from any process that shares the agent's PubSub. Only the runId from stream() or prepare() is needed, so a later request can stop a run it didn't start. The run ends the same way as abort() on the stream result: the stream emits an abort chunk, onAbort fires, and observe() consumers see the stream end.
const { runId } = await durableAgent.stream('Summarize the document')
// later, from a different request handler
durableAgent.abortRunStream(runId)
Publishing the abort request over PubSub requires a shared backend in a multi-process deployment, such as RedisStreamsPubSub. The default EventEmitterPubSub only reaches runs in the same process. A run parked on a tool approval is cancelled only in this process: another process sharing the storage can still approve it. To cancel a parked run on every server, call the server's POST /agents/:agentId/runs/:runId/abort route, which records the cancellation in storage so a later approval or resume is refused with 409 Conflict (AGENT_RESUME_ABORTED_RUN).
InngestAgent doesn't implement this method. Cancel an Inngest run with the abort() returned by its stream().
Stopping a durable run through abortRunStream() or abortThreadStream() requires @mastra/core@1.62.0 or later. Earlier versions record the abort without stopping the run.
Returns: boolean. true when this process aborted the run locally or can see it executing. The abort request is published either way.
abortThreadStream({ threadId, resourceId?, expectedRunId? })Direct link to abortthreadstream-threadid-resourceid-expectedrunid-
Aborts the active run on a memory thread with the same abort request as abortRunStream(). The run is resolved from this process's thread runtime, so it must have been started here or observed through subscribeToThread() on this process. The server route POST /agents/:agentId/threads/abort uses this method.
Pass expectedRunId when the request must only stop a specific run. If another queued run becomes active before the request is handled, the method returns false without aborting the successor. Omit expectedRunId to abort whichever run is active when the request is handled.
const aborted = durableAgent.abortThreadStream({
resourceId: 'user-1',
threadId: 'thread-1',
expectedRunId: runId,
})
Returns: boolean. false when this process has no active run recorded for the thread or the active run doesn't match expectedRunId. No abort request is sent in either case.
RecoveryDirect link to Recovery
Recovery re-drives a run from its last persisted snapshot. Mastra treats a sub-agent delegation as one generated agent-<name> tool step and doesn't checkpoint its inner LLM or tool calls independently. If recovery replays that step, the entire sub-agent run can execute again, including already-completed tool side effects. Make tools used by sub-agents idempotent, or deduplicate their external side effects. See crash recovery for operational guidance.
listActiveRuns(options?)Direct link to listactiverunsoptions
Lists this agent's runs whose persisted snapshot is in running status: runs whose agentic loop was mid-execution when the workflow engine last saved state. On a live process they transition to suspended or a terminal status. After a crash or restart they stay running with nothing driving them, which is what recoverActiveRuns() re-drives. Runs started by other durable agents on the same storage aren't included.
running snapshots are only written when recovery.durableAgents is 'auto' or a custom shouldPersistSnapshot includes running. Without one of those, this method returns no runs. See snapshot persistence.
const { runs, total } = await durableAgent.listActiveRuns({ resourceId: 'user-1' })
for (const run of runs) {
await durableAgent.recoverActiveRuns({ runId: run.runId })
}
Reads workflow snapshot storage on the Mastra instance the agent is registered with, and throws if the agent isn't registered with one. Use persistent storage: the default in-memory store loses these rows on the restart that orphans them.
Only runs in running status are returned. Suspended runs keep their snapshots but aren't included, and finished runs aren't retained. Filter by resourceId to scope the list to one user.
Returns:
interface DurableAgentListActiveRunsResult {
runs: Array<{
runId: string
status: 'running'
threadId?: string
resourceId?: string
updatedAt: Date
}>
total: number
}
options.threadId?:
options.resourceId?:
options.fromDate?:
options.toDate?:
options.perPage?:
page is also set. Must be a positive integer.options.page?:
perPage is also set. Must be a non-negative integer.recoverActiveRuns(options?)Direct link to recoveractiverunsoptions
Discovers runs stuck in running status for this agent and re-drives them from the last persisted snapshot. Accepts the same filters and pagination as listActiveRuns(), plus runId. Returns the outcome for each run.
Up to five run waits proceed concurrently, with results returned in discovery order. Each attempt has a single 30-second wait budget covering both recover() setup and settlement of the whole recovered workflow. When the budget expires, another run can use the slot. The timeout returns status: 'skipped' with reason: 'recovery-timeout' and retryAt set 30 seconds ahead, without canceling setup or execution, releasing a live claim, or counting as success or failure. The attempt continues in the background, and its eventual outcome doesn't change the returned results. The concurrency limit bounds waiting, not live executions.
mastra.recoverAllDurableAgents() schedules timeout re-checks automatically. Direct callers of recoverActiveRuns() must schedule their own calls after retryAt. While the original bulk attempt remains pending, re-checks defer with another recovery-timeout and a future retryAt, without invoking recover() again. Once it settles, recovery checks the persisted snapshot and ownership before retrying. Finished or suspended runs aren't restarted.
The budget applies only to each run attempt in recoverActiveRuns(), not to discovery, a direct recover() call, or overridden custom recovery methods. When shutdown begins, recovery stops dispatching new runs. The original attempt remains tracked for Mastra's shared shutdown drain even after the wait times out.
const result = await durableAgent.recoverActiveRuns()
// {
// recovered: [
// { runId: 'run-1', status: 'success' },
// { runId: 'run-2', status: 'skipped', reason: 'run-active', retryAt: 1760123456789 },
// ],
// succeeded: 1,
// failed: 0,
// }
Pass a runId to recover a single known run:
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })
For a run that another execution still drives, recoverActiveRuns() returns status: 'skipped' and a reason, leaving that execution in control. Checking the run again requires a separate call. For a run held by another process, retryAt is the time after which that claim lapses unless renewed. Call recoverActiveRuns({ runId }) after it to recover the run if its process crashed. mastra.recoverAllDurableAgents() schedules these re-checks itself.
reason | Meaning |
|---|---|
run-active | Another execution, possibly in another process, holds the run. Sets retryAt. |
run-active-locally | This process is executing the run. |
already-in-progress | This process is already recovering the run. |
recovery-timeout | The wait expired or an earlier bulk recovery attempt is still pending. Sets retryAt without duplicating the pending attempt on re-checks. |
Returns:
interface DurableAgentRecoverActiveRunsResult {
recovered: Array<{
runId: string
status: 'success' | 'failed' | 'skipped'
error?: Error
reason?: 'run-active' | 'run-active-locally' | 'already-in-progress' | 'recovery-timeout'
// Epoch milliseconds
retryAt?: number
}>
succeeded: number
failed: number
}
options.runId?:
recover(runId, options?)Direct link to recoverrunid-options
Recovers a single run by ID. Returns a streamable result with the same shape as stream(). Use this when you need to observe the recovery stream in real time.
const { output, cleanup } = await durableAgent.recover('run-abc-123', {
onChunk: chunk => console.log(chunk),
onError: ({ error }) => console.error(error),
})
await output.text
cleanup()
Recovery restarts from the persisted durable-agentic-loop checkpoint. The nested execution row alone isn't sufficient. Before starting a run, enable recovery: { durableAgents: 'auto' } on Mastra, or provide a shouldPersistSnapshot policy that includes 'running'. Recovery is unavailable until the first running checkpoint is saved, even with persistence enabled. Enabling persistence after a crash can't reconstruct a missing checkpoint.
Model and tool calls after the checkpoint may run again. Make tool side effects idempotent. Recovery doesn't guarantee exactly-once execution.
recover() throws for unavailable checkpoints and runs that aren't orphaned:
| Error ID | Thrown when |
|---|---|
DURABLE_AGENT_RECOVER_CHECKPOINT_NOT_AVAILABLE | The loop row is still pending without a workflow input. No restartable checkpoint is available. No recovered execution starts, and any acquired recovery claim is released. |
DURABLE_AGENT_RUN_ACTIVE | Another execution holds the run, and force isn't set. |
DURABLE_AGENT_RECOVER_RUN_ACTIVE_LOCALLY | This process is executing the run. force doesn't override this. |
DURABLE_AGENT_RECOVER_ALREADY_IN_PROGRESS | This process is already recovering the run. |
DURABLE_AGENT_RECOVER_RUN_SUSPENDED | The run is suspended. Continue it with resume(). |
DURABLE_AGENT_RECOVER_RUN_NOT_ACTIVE | The snapshot isn't restartable and isn't suspended. |
DURABLE_AGENT_EXECUTION_CONFLICT | With force, another execution claimed the run during the takeover. |
The server's POST /agents/:agentId/recover route returns ownership, already-in-progress, suspended-run, and inactive-run conflicts as 409 Conflict. An unavailable checkpoint returns 500 Internal Server Error.
options.force?:
Returns: Promise<DurableAgentStreamResult>
Stream optionsDirect link to Stream options
stream() accepts a DurableAgentStreamOptions object. It supports the agent execution options below, plus lifecycle callbacks.
runId?:
resume(), observe(), or abortRunStream().instructions?:
context?:
memory?:
requestContext?:
maxSteps?:
toolsets?:
clientTools?:
toolChoice?:
activeTools?:
modelSettings?:
Authorization, X-Api-Key, and similar) are stripped from the serialized snapshot before it crosses process boundaries.stopWhen?:
fullStream rejects without emitting a step-finish or error chunk. Durable transports preserve the error name and message, but not its stack, cause, or subclass identity. The closure rides on the in-process run registry; cross-process resumes degrade to maxSteps only.system?:
requireToolApproval?:
true or false to gate all or none, or a function for per-call policy. Function-form policies live on the in-process run registry; cross-process resumes fall back to a true shadow.autoResumeSuspendedTools?:
resume() call.closeOnSuspend?:
true, close the stream when the run suspends (for example for tool approval) so fullStream, text, and getFullOutput() resolve at the suspension boundary instead of staying open, matching non-durable Agent.stream(). Useful for callers such as AG-UI or A2A that need the stream to end. Defaults to false, which keeps the stream open across suspension so a later resume can continue streaming on the same reader. resume()/resumeStream() always return a fresh stream.toolCallConcurrency?:
includeRawChunks?:
maxProcessorRetries?:
structuredOutput?:
untilIdle?:
true for the default 5-minute idle timeout, or { maxIdleMs } to customise. Equivalent to the deprecated streamUntilIdle() method. Also supported on resume().disableBackgroundTasks?:
tracingOptions?:
requestContextKeys forwarded to the agent and model spans. Fully JSON-serializable.actor?:
transform?:
transformToolPayload closure lives on the in-process run registry; only the JSON-safe targets shadow is serialized. Recovery after a process restart cannot restore the closure. For crash-survivable transcript transforms, configure transform.transcript on the tool with createTool() so the durable worker can re-resolve it from the registered tool.prepareStep?:
PrepareStepProcessor at the start of every iteration. Closure-only — stored on the in-process run registry. Cross-process resumes lose the hook.isTaskComplete?:
onComplete live on the in-process run registry; the JSON-safe primitives (strategy, timeout, parallel, suppressFeedback, scorerNames) are serialized for cross-process observability.delegation?:
onDelegationStart, onDelegationComplete, messageFilter). Callbacks passed per call are baked into the sub-agent tool wrappers at prepare time, and cross-process resumes lose them, so a fresh worker falls back to default delegation behavior. Hooks configured in the agent's defaultOptions.delegation are merged back in when recovery rebuilds tools, so put hooks that must survive a restart there.versions?:
abortSignal?:
AbortController, so either source can cancel the run. Cross-process resumes cannot recover the signal — pass a fresh one to resume() if you need post-resume abortability.onChunk?:
onStepFinish?:
onFinish?:
onAbort; failed runs call onError.onError?:
onSuspended?:
onAbort?:
abortSignal, result.abort(), abortRunStream(), or abortThreadStream(). Receives the steps completed before the abort and, in text, the assistant text streamed so far.onIterationComplete?:
messageList, finishReason, and isFinal flag. Return continue: true to request another iteration or continue: false to stop. Feedback is added as a synthetic assistant message when the loop runs another iteration. If the model has stopped, feedback alone or continue: false with feedback does not start another iteration. When the loop was already going to continue, continue: false with feedback runs one final feedback-guided iteration before stopping.resume(), observe(), and recover() accept the same lifecycle callbacks (onChunk, onStepFinish, onFinish, onError, onAbort, onSuspended). observe() also accepts an offset to control where replay starts.
DurableAgentStreamResultDirect link to DurableAgentStreamResult
The object returned by stream(), resume(), observe(), and recover().
interface DurableAgentStreamResult<OUTPUT = undefined> {
output: MastraModelOutput<OUTPUT>
readonly fullStream: ReadableStream<any>
runId: string
threadId?: string
resourceId?: string
cleanup: () => void
abort: () => void
}
output:
output.text for the full text, or consume output.fullStream.fullStream:
output.fullStream.runId:
resume() or observe() to reconnect.threadId?:
resourceId?:
cleanup:
abort:
AbortController. Surfaces as an AbortError inside the durable LLM-execution step and fires the onAbort callback. Safe to call after the run has finished — a no-op in that case.