import type { MastraServerCache } from '../../cache/base.js'; import type { PubSub } from '../../events/pubsub.js'; import type { Mastra } from '../../mastra/index.js'; import type { FullOutput, MastraModelOutput } from '../../stream/base/output.js'; import type { ChunkType, MastraOnFinishCallback } from '../../stream/types.js'; import type { WorkflowRunStatus } from '../../workflows/types.js'; import { Agent } from '../agent.js'; import type { AgentExecutionOptions } from '../agent.types.js'; import type { MessageListInput } from '../message-list/index.js'; import type { ToolsInput } from '../types.js'; import { ExtendedRunRegistry } from './run-registry.js'; import type { AgentStepFinishEventData, AgentSuspendedEventData, DurableAgenticWorkflowInput } from './types.js'; import { createDurableAgenticWorkflow } from './workflows/index.js'; /** * Options for DurableAgent.stream() */ export interface DurableAgentStreamOptions { /** Custom instructions that override the agent's default instructions for this execution */ instructions?: AgentExecutionOptions['instructions']; /** Additional context messages to provide to the agent */ context?: AgentExecutionOptions['context']; /** Memory configuration for conversation persistence and retrieval */ memory?: AgentExecutionOptions['memory']; /** Unique identifier for this execution run */ runId?: string; /** Request Context containing dynamic configuration and state */ requestContext?: AgentExecutionOptions['requestContext']; /** Maximum number of steps to run */ maxSteps?: number; /** * Conditions for stopping execution (e.g., step count, token limit). * * The predicate is non-serializable, so it's parked on the in-process run * registry and evaluated by the durable loop on every iteration. Cross-process * durable engines (e.g. Inngest after a worker restart) cannot recover the * closure and degrade to `maxSteps` only. */ stopWhen?: AgentExecutionOptions['stopWhen']; /** Additional tool sets that can be used for this execution */ toolsets?: AgentExecutionOptions['toolsets']; /** Client-side tools available during execution */ clientTools?: AgentExecutionOptions['clientTools']; /** Tool selection strategy */ toolChoice?: AgentExecutionOptions['toolChoice']; /** Tool names enabled for this execution */ activeTools?: AgentExecutionOptions['activeTools']; /** Model-specific settings like temperature */ modelSettings?: AgentExecutionOptions['modelSettings']; /** Require approval for tool calls. Boolean (gate all / none) or a per-call function policy. */ requireToolApproval?: AgentExecutionOptions['requireToolApproval']; /** Automatically resume suspended tools */ autoResumeSuspendedTools?: boolean; /** Maximum number of tool calls to execute concurrently */ toolCallConcurrency?: number; /** Whether to include raw chunks in the stream output */ includeRawChunks?: boolean; /** Maximum processor retries */ maxProcessorRetries?: number; /** Structured output configuration */ structuredOutput?: AgentExecutionOptions['structuredOutput']; /** Version overrides for sub-agent delegation */ versions?: AgentExecutionOptions['versions']; /** Callback when chunk is received */ onChunk?: (chunk: ChunkType) => void | Promise; /** Callback when step finishes */ onStepFinish?: (result: AgentStepFinishEventData) => void | Promise; /** Callback when execution finishes — receives rich step data (text, steps, toolResults) */ onFinish?: MastraOnFinishCallback; /** Callback on error */ onError?: ({ error }: { error: Error | string; }) => void | Promise; /** Callback when workflow suspends (e.g., for tool approval) */ onSuspended?: (data: AgentSuspendedEventData) => void | Promise; /** Callback when execution is aborted via abortSignal */ onAbort?: AgentExecutionOptions['onAbort']; /** Callback fired after each agentic-loop iteration */ onIterationComplete?: AgentExecutionOptions['onIterationComplete']; /** Additional system message appended after context but before user messages. */ system?: AgentExecutionOptions['system']; /** When true, background tasks are disabled for this run. */ disableBackgroundTasks?: AgentExecutionOptions['disableBackgroundTasks']; /** Tracing options forwarded to the agent/model spans. */ tracingOptions?: AgentExecutionOptions['tracingOptions']; /** Per-call actor signal forwarded to FGA checks and tool execution. */ actor?: AgentExecutionOptions['actor']; /** * Per-invocation tool payload transform policy. The closure rides on the * in-process run registry; only the JSON-safe `targets` shadow is serialized * for cross-process engines. */ transform?: AgentExecutionOptions['transform']; /** * Per-step preparation hook. Closure-only: stored on the in-process run * registry and invoked as a `PrepareStepProcessor` at the start of every * iteration. Cross-process resumes lose the hook. */ prepareStep?: AgentExecutionOptions['prepareStep']; /** * Per-call `isTaskComplete` policy. Scorer instances and `onComplete` are * closure-only and live on the in-process run registry; the JSON-safe * primitives (`strategy`, `timeout`, `parallel`, `suppressFeedback`, * `scorerNames`) are serialized for cross-process observability. */ isTaskComplete?: AgentExecutionOptions['isTaskComplete']; /** * Sub-agent delegation hooks (`onDelegationStart`, `onDelegationComplete`, * `messageFilter`, etc.). The callbacks are forwarded into `convertTools` * at prepare time and burned into the sub-agent `CoreTool` wrappers on the * in-process run registry. Cross-process resumes lose the callbacks (only * `includeSubAgentToolResultsInModelContext` would be JSON-safe), so a * fresh worker degrades to default delegation behaviour. */ delegation?: AgentExecutionOptions['delegation']; /** * When set, `stream()` delegates to the idle-loop wrapper that keeps the * outer stream open across background-task continuations — the same * behaviour as the now-deprecated `streamUntilIdle()`. * * Pass `true` for default idle timeout (5 min), or `{ maxIdleMs }` to * customise. * * @example * ```typescript * const { output, cleanup } = await durableAgent.stream('Research topic', { * untilIdle: true, * memory: { thread: 't1', resource: 'u1' }, * }); * ``` */ untilIdle?: boolean | { maxIdleMs?: number; }; /** When true, the in-loop background task check step skips waiting (streamUntilIdle sets this) */ _skipBgTaskWait?: boolean; /** * External abort signal. The durable agent always installs its own internal * `AbortController` for the run; when this signal is provided, its `abort` * event is forwarded to the internal controller so either source can cancel * the run. * * Cross-process resumes (e.g. Inngest after a worker restart) cannot * recover the signal — call `resume(runId, ..., { abortSignal })` with a * fresh signal on each segment if you need abortability post-resume. */ abortSignal?: AbortSignal; } /** * Result from DurableAgent.stream() */ export interface DurableAgentStreamResult { /** The streaming output */ output: MastraModelOutput; /** The full stream - delegates to output.fullStream for server compatibility */ readonly fullStream: ReadableStream; /** The unique run ID for this execution */ runId: string; /** Thread ID if using memory */ threadId?: string; /** Resource ID if using memory */ resourceId?: string; /** Cleanup function to call when done (unsubscribes from pubsub) */ cleanup: () => void; /** * Abort the run. Flips the internal `AbortController` for this run, which * surfaces as an `AbortError` inside the durable LLM-execution step and * is bridged to the user's `onAbort` callback via the run's pubsub topic. * * Safe to call after the run has already finished — it's a no-op in that * case. */ abort: (reason?: unknown) => void; } /** * Configuration for DurableAgent - wraps an existing Agent with durable execution */ export interface DurableAgentConfig { /** * The Agent to wrap with durable execution capabilities. * All agent methods (getModel, listTools, etc.) delegate to this agent. */ agent: Agent; /** * Optional ID override. Defaults to agent.id. */ id?: TAgentId; /** * Optional name override. Defaults to agent.name. */ name?: string; /** * PubSub instance for streaming events. * Optional - if not provided, defaults to EventEmitterPubSub. */ pubsub?: PubSub; /** * Cache instance for storing stream events. * Enables resumable streams - clients can disconnect and reconnect * without missing events. * * - If not provided: Inherits from Mastra instance, or uses InMemoryServerCache * - If provided: Uses the provided cache backend (e.g., Redis) * - If set to `false`: Disables caching (streams are not resumable) */ cache?: MastraServerCache | false; /** * Maximum steps for the agentic loop. * Defaults to the workflow default if not specified. */ maxSteps?: number; /** * Timeout in milliseconds before automatic cleanup of registry entries * after a stream finishes or errors. This provides a grace period for * late observers to access the stream. * * Defaults to 30000 (30 seconds). * Set to 0 to disable auto-cleanup (manual cleanup() required). */ cleanupTimeoutMs?: number; } /** * DurableAgent wraps an existing Agent with durable execution capabilities. * * Key features: * 1. Resumable streams - clients can disconnect and reconnect without missing events * 2. Serializable workflow inputs - works with durable execution engines * 3. PubSub-based streaming - events flow through pubsub for distribution * * DurableAgent extends Agent, delegating most methods to the wrapped agent. * It overrides stream() to use durable execution with the agentic workflow. * * Subclasses (EventedAgent, InngestAgent) override executeWorkflow() to * customize how the workflow is executed. * * @example * ```typescript * import { Agent } from '@mastra/core/agent'; * import { DurableAgent } from '@mastra/core/agent/durable'; * * const agent = new Agent({ * id: 'my-agent', * instructions: 'You are a helpful assistant', * model: openai('gpt-4'), * }); * * const durableAgent = new DurableAgent({ agent }); * * const { output, runId, cleanup } = await durableAgent.stream('Hello!'); * const text = await output.text; * cleanup(); * ``` */ /** * Statuses of durable agent runs discoverable via {@link DurableAgent.listActiveRuns}. * * `running` is the status reported by the workflow engine while the durable * agent's agentic loop is actively executing (i.e. between suspend * boundaries). Persisted `running` snapshots are the recovery source for runs * orphaned by a process restart. */ export type DurableAgentActiveRunStatus = Extract; /** * Filters for {@link DurableAgent.listActiveRuns}. Mirrors the * `listWorkflowRuns` filter contract, plus the agent-level `threadId` / * `resourceId` filters used by the base {@link Agent.listSuspendedRuns}. */ export interface DurableAgentListActiveRunsOptions { /** Only return runs that belong to this memory thread. */ threadId?: string; /** Only return runs that belong to this memory resource. */ resourceId?: string; /** Only return runs created at or after this date. */ fromDate?: Date; /** Only return runs created at or before this date. */ toDate?: Date; /** * Number of items per page. Pagination is applied when both `perPage` and * `page` are provided; otherwise all matching runs are returned. */ perPage?: number; /** Zero-indexed page number. */ page?: number; } /** * A durable agent run currently reported as `running` in workflow snapshot * storage. These are the runs that a boot-time or operator-initiated * recovery would re-drive after a process restart. */ export interface DurableAgentActiveRun { /** Run ID accepted by {@link DurableAgent.recoverActiveRuns} and workflow `restart`. */ runId: string; status: DurableAgentActiveRunStatus; threadId?: string; resourceId?: string; /** When the run's snapshot was last persisted while running. */ updatedAt: Date; } export interface DurableAgentListActiveRunsResult { runs: DurableAgentActiveRun[]; /** Total number of matching runs, before pagination. */ total: number; } /** * Outcome of a single run restart attempted by * {@link DurableAgent.recoverActiveRuns}. `success` means `run.restart()` * returned; `failed` means it threw and the error was captured so recovery * of remaining runs could proceed. */ export interface DurableAgentRecoveredRun { runId: string; status: 'success' | 'failed'; /** Populated only when `status === 'failed'`. */ error?: Error; } /** * Filters for {@link DurableAgent.recoverActiveRuns}. Reuses the * {@link DurableAgentListActiveRunsOptions} discovery filters and adds an * escape hatch for targeting a specific run ID. */ export interface DurableAgentRecoverActiveRunsOptions extends DurableAgentListActiveRunsOptions { /** * Recover a specific run by ID. When set, the discovery filters and * pagination fields are ignored. Useful when the caller already knows the * run ID from another source (e.g. their own bookkeeping). */ runId?: string; } export interface DurableAgentRecoverActiveRunsResult { recovered: DurableAgentRecoveredRun[]; /** Number of runs that restarted successfully. */ succeeded: number; /** Number of runs whose restart threw. */ failed: number; } /** * Options for {@link DurableAgent.recover}, a single-run streamable recovery * counterpart to {@link DurableAgent.resume}. * * `recover()` rebuilds the run's non-serializable state from the persisted * workflow snapshot (message list, model, tools, memory, saveQueueManager, * request context, agent span) and returns a fresh {@link DurableAgentStreamResult} * whose `fullStream` observes the recovered run through pubsub. Callbacks * mirror `stream()` / `resume()`. */ export interface DurableAgentRecoverOptions { /** Callback when chunk is received */ onChunk?: (chunk: ChunkType) => void | Promise; /** Callback when a step finishes */ onStepFinish?: (result: AgentStepFinishEventData) => void | Promise; /** Callback when the recovered run finishes */ onFinish?: MastraOnFinishCallback; /** Callback when the recovered run errors */ onError?: ({ error }: { error: Error | string; }) => void | Promise; /** Callback when the recovered run suspends again */ onSuspended?: (data: AgentSuspendedEventData) => void | Promise; /** * Optional abort signal for the recovered segment. Forwarded onto a fresh * internal `AbortController` installed on the run's registry entry, so * `result.abort()` and the external signal can both cancel the recovered run. */ abortSignal?: AbortSignal; } export declare class DurableAgent extends Agent { #private; /** * Create a new DurableAgent that wraps an existing Agent */ constructor(config: DurableAgentConfig); /** * Get the resolved cache instance. * Lazily initialized to allow inheriting from Mastra. */ get cache(): MastraServerCache | null; /** * Get the PubSub instance. * Returns CachingPubSub if caching is enabled, otherwise the inner pubsub. */ get pubsub(): PubSub; /** * Get the wrapped agent instance. */ get agent(): Agent; /** * Get the run registry (for testing and advanced usage) */ get runRegistry(): ExtendedRunRegistry; /** * Get the max steps configured for this agent */ get maxSteps(): number | undefined; /** * Get the cleanup timeout in milliseconds. * Returns 0 if auto-cleanup is disabled. */ get cleanupTimeoutMs(): number; getModel(options?: any): import("../../_types/@internal_ai-sdk-v4/dist/index.d.ts").LanguageModelV1 | import("..").MastraLanguageModel | Promise; getLLM(options?: any): import("..").MastraLLM | Promise; getModelList(requestContext?: any): Promise; getInstructions(options?: any): import("../../llm").SystemMessage | Promise; getDescription(): string; getMetadata(options?: any): Record | Promise | undefined> | undefined; getTracingPolicy(): import("../../observability").TracingPolicy | undefined; listTools(options?: any): TTools | Promise; getConfiguredToolHooks(): import("../../tools").ToolHooks> | undefined; getDefaultOptions(options?: any): AgentExecutionOptions | Promise>; getDefaultGenerateOptionsLegacy(options?: any): import("..").AgentGenerateOptions | Promise; getDefaultStreamOptionsLegacy(options?: any): import("..").AgentStreamOptions | Promise; getDefaultNetworkOptions(options?: any): import("..").NetworkOptions | Promise; getMemory(options?: any): Promise; hasOwnMemory(): boolean; getWorkspace(options?: any): Promise; hasOwnWorkspace(): boolean; getVoice(options?: any): Promise>; get voice(): import("../../_types/@internal_voice/dist/index.d.ts").MastraVoice; get requestContextSchema(): import("@mastra/schema-compat").StandardSchemaWithJSON | undefined; getConfiguredProcessorWorkflows(): Promise; listInputProcessors(requestContext?: any): Promise; listOutputProcessors(requestContext?: any): Promise; listErrorProcessors(requestContext?: any): Promise; resolveProcessorById(processorId: TId, requestContext?: any): Promise | null>; listConfiguredInputProcessors(requestContext?: any): Promise; listConfiguredOutputProcessors(requestContext?: any): Promise; getConfiguredProcessorIds(requestContext?: any): Promise<{ inputProcessorIds: string[]; outputProcessorIds: string[]; errorProcessorIds: string[]; }>; listAgents(options?: any): Record> | Promise>>; __getStaticAgents(): Record> | undefined; __hasSubAgentsConfigured(): boolean; listWorkflows(options?: any): Promise>; getSkill(skillName: string, options?: any): Promise; listSkills(options?: any): Promise; listScorers(options?: any): Promise; getBackgroundTasksConfig(): import("../../background-tasks").AgentBackgroundConfig | undefined; disableBackgroundTasks(): void; enableBackgroundTasks(): void; getToolPayloadTransform(): import("../../tools").ToolPayloadTransformPolicy | undefined; __getGoalConfig(): import("..").GoalConfig | undefined; get browser(): import("..").MastraBrowser | undefined; setBrowser(browser: any): void; hasOwnBrowser(): boolean; getChannels(): import("../../channels").AgentChannels | null; setChannels(agentChannels: any): void; hasOwnPubSub(): boolean; __setMemory(memory: any): void; __setPubSub(pubsub: any): void; __setWorkspace(workspace: any): void; __getEditorConfig(): import("..").AgentEditorConfig | undefined; __getOverridableFields(): { instructions: import("../../types").DynamicArgument; model: { id: string; model: import("../../types").DynamicArgument; maxRetries: number; enabled: boolean; modelSettings?: import("../../types").DynamicArgument; providerOptions?: import("../../types").DynamicArgument; headers?: import("../../types").DynamicArgument>; }[] | import("../../types").DynamicArgument; tools: import("../../types").DynamicArgument; workspace: import("../../types").DynamicArgument; }; __updateInstructions(instructions: Parameters['__updateInstructions']>[0]): void; __updateModel(config: Parameters['__updateModel']>[0]): void; __setTools(tools: Parameters['__setTools']>[0]): void; /** * Create a per-request clone for applying stored editor overrides. * * The base `Agent.__fork()` builds a bare `new Agent(...)`, which for a * DurableAgent would drop the wrapped agent and every delegating override * (tools, model, memory, voice, durable streaming) — the served fork ends up a * plain `Agent` with no tools. Instead, fork the wrapped agent (so overrides * applied to this fork don't mutate the singleton) and re-wrap it in the same * durable subclass, preserving pubsub/cache/run configuration. * * @internal */ __fork(): Agent; /** * Get the PubSub instance for use by subclasses. * @internal */ protected get pubsubInternal(): PubSub; /** * Get the run registry for use by subclasses. * @internal */ protected get runRegistryInternal(): ExtendedRunRegistry; /** * Execute the durable workflow. * * Subclasses override this method to customize how the workflow is executed: * - DurableAgent (this): Runs the workflow directly via createRun + start * - EventedAgent: Uses run.startAsync() for fire-and-forget execution * - InngestAgent: Uses inngest.send() to trigger Inngest function * * @param runId - The unique run ID * @param workflowInput - The serialized workflow input * @internal */ protected executeWorkflow(runId: string, workflowInput: DurableAgenticWorkflowInput): Promise; /** * Create the durable workflow for this agent. * * Subclasses can override this method to use a different workflow implementation: * - DurableAgent (this): Uses createDurableAgenticWorkflow() * - InngestAgent: Uses createInngestDurableAgenticWorkflow() * * @internal */ protected createWorkflow(): ReturnType; /** * Emit an error event to pubsub. * * @param runId - The run ID * @param error - The error to emit * @internal */ protected emitError(runId: string, error: Error): Promise; /** * Delete the persisted workflow snapshot rows for a completed durable run. * * A durable agent write two rows per run: one for the outer `AGENTIC_LOOP` * workflow and one for the nested `AGENTIC_EXECUTION` workflow (persisted * under the same `runId`). Once the run reaches a non-suspended terminal * state neither row is needed again — leaving them behind fills snapshot * storage with stale `pending`/`running` rows for every completed run and * pollutes `listActiveRuns` / `recoverActiveRuns` on the next boot. * * Best-effort: a cleanup failure must never turn a finished run into an * error — a stale row is preferable to a broken exit path. * * @internal */ protected deleteRunSnapshots(runId: string): Promise; /** * Stream a response from the agent using durable execution. */ stream(messages: MessageListInput, options?: DurableAgentStreamOptions): Promise>; /** * Resume a suspended workflow execution. */ resume(runId: string, resumeData: unknown, options?: DurableAgentStreamOptions): Promise>; /** * Recover a single durable run whose in-process agentic loop was orphaned by * a process restart. Streamable counterpart to * {@link DurableAgent.recoverActiveRuns} — where the bulk API only re-drives * the workflow and returns counts, `recover()` rebuilds the run's * non-serializable state (message list, model, tools, memory, * saveQueueManager, request context, agent span) from the persisted workflow * snapshot and returns a fresh {@link DurableAgentStreamResult} whose * `fullStream` observes the recovered run through pubsub. * * Because the rebuilt registry entry carries `memory` + `saveQueueManager`, * the durable agentic workflow's terminal step will flush new messages to * memory just like a fresh `stream()` call would. The single-run form is * useful when operators want to attach listeners to a specific recovered * run; for boot-time bulk recovery of every orphaned run, use * `recoverActiveRuns()`. * * @example * ```typescript * const { fullStream, output, cleanup } = await durableAgent.recover(runId, { * onChunk: chunk => process.stdout.write(chunk.payload?.text ?? ''), * }); * for await (const chunk of fullStream) { * // ... * } * cleanup(); * ``` */ recover(runId: string, options?: DurableAgentRecoverOptions): Promise>; /** * Override the inherited `resumeStream()` so that callers using the base * `Agent` API (including `approveToolCall` / `declineToolCall`) are routed * through the durable `resume()` path instead of the regular Agent's * snapshot-based resume. * * Returns just the `MastraModelOutput` (matching the base Agent's return * type) while internally delegating to `this.resume()`. */ resumeStream(resumeData: any, streamOptions?: any): Promise>; /** * Override the inherited `approveToolCall()` to route through the durable * `resume()` path. */ approveToolCall(options: { runId: string; toolCallId?: string; } & Record): Promise>; /** * Override the inherited `declineToolCall()` to route through the durable * `resume()` path. */ declineToolCall(options: { runId: string; toolCallId?: string; } & Record): Promise>; /** * Generate a complete response from the agent using durable execution. * * Drains the underlying durable stream to completion and returns the same * {@link FullOutput} shape as non-durable `Agent.generate`. The underlying * workflow is identical to `stream()` — it just collects the final result * for callers that don't want to consume chunks themselves. * * This method intentionally re-implements the `stream()` setup rather than * delegating to `this.stream(...)` so that `prepareForDurableExecution` (and * downstream `convertTools`) receives `methodType: 'generate'`. Tool * factories that vary their `CoreTool` output based on the calling method * (e.g. `clientTools` vs server-side tools) rely on this signal — calling * `stream()` here would silently pass `methodType: 'stream'`. * * If the run suspends (e.g. tool approval or `suspend()` from a tool), the * returned output's `finishReason` will be `'suspended'` and * `suspendPayload` will be populated. Use {@link DurableAgent.resumeGenerate} * to continue. * * Note on suspend persistence: for the base `DurableAgent`, the workflow * engine's `run.start()` only resolves after the suspend snapshot is * persisted, so awaiting `workflowExecution` on suspend is sufficient for * a subsequent `resumeGenerate()` to find the snapshot. Subclasses like * `EventedAgent` use a fire-and-forget `run.startAsync()` and therefore * cannot rely on this await for snapshot durability — see the * `EventedAgent` docs for the recommended pattern. */ generate(messages: MessageListInput, options?: DurableAgentStreamOptions): Promise>; /** * Resume a suspended durable run and drain it to a single * {@link FullOutput}. Mirrors {@link Agent.resumeGenerate} on top of * {@link DurableAgent.resume}. * * Unlike `generate()`, this delegates to `resume()` because resume reads * its tools from the existing run-registry entry rather than running * `prepareForDurableExecution` again — there is no `methodType` to thread * through. The same `EventedAgent` caveat about fire-and-forget snapshot * persistence noted on `generate()` applies if the resumed turn suspends. */ resumeGenerate(runId: string, resumeData: unknown, options?: Parameters['resume']>[2]): Promise>; /** * List durable agent runs currently reported as `running` in workflow * snapshot storage. * * A `running` snapshot is a durable agent run whose agentic loop was * mid-execution the last time the workflow engine persisted its state. On a * healthy process these transition to `suspended` (waiting on * tool approval / resume) or a terminal status. On a crashed / restarted * process they are orphaned in the `running` state with no in-process * driver — this is the discovery API used to enumerate them for recovery * (see {@link DurableAgent.recoverActiveRuns} and workflow `restart`). * * Requires persistent workflow storage. Filters `agentId` against the * persisted `DurableAgenticWorkflowInput.agentId`, so runs started by other * durable agents sharing the same storage are not surfaced. * * @example * ```typescript * const { runs } = await durableAgent.listActiveRuns({ resourceId }); * for (const run of runs) { * await durableAgent.recoverActiveRuns({ runId: run.runId }); * } * ``` */ listActiveRuns(options?: DurableAgentListActiveRunsOptions): Promise; /** * Bulk recover durable agent runs whose in-process agentic loop was orphaned * by a process restart. This is the recovery half of the discovery API * paired with {@link DurableAgent.listActiveRuns} and is the typical * boot-time hook. * * Each targeted run is delegated to {@link DurableAgent.recover}, which * rebuilds the run's non-serializable state (message list, model, memory, * save-queue manager, request context, agent span), re-subscribes to the * run's pubsub topic, and restarts the workflow in the background. Because * `recover()` registers `memory` + `saveQueueManager` on the run entry, the * durable agentic workflow's terminal step flushes new messages to memory * just like a fresh `stream()` call would. * * The per-run stream returned by `recover()` is discarded — this method * awaits each run's workflow settlement and reports summary counts instead * of surfacing live event streams. Callers who want to observe a specific * recovered run's events should use {@link DurableAgent.recover} directly * (or {@link DurableAgent.observe} with the returned `runId`). * * Failures are captured per-run so a single bad run does not block * recovery of the rest. * * @example * ```typescript * // Recover every orphaned run for this agent (typical boot-time hook). * const { recovered, succeeded, failed } = await durableAgent.recoverActiveRuns(); * logger.info('Recovered durable agent runs', { succeeded, failed }); * * // Recover a single run by ID. * await durableAgent.recoverActiveRuns({ runId }); * ``` */ recoverActiveRuns(options?: DurableAgentRecoverActiveRunsOptions): Promise; /** * Observe an existing stream. * Use this to reconnect to a stream after a network disconnection. * * **Warning:** The returned `cleanup()` function destroys the run's registry * entries and cached PubSub events. Only call it when you are done with the * run entirely. If the workflow is suspended and you intend to resume later, * do not call cleanup — let the auto-cleanup timer handle it after * FINISH/ERROR. Auto-cleanup does not fire on SUSPENDED events. */ observe(runId: string, options?: { offset?: number; onChunk?: (chunk: ChunkType) => void | Promise; onStepFinish?: (result: AgentStepFinishEventData) => void | Promise; onFinish?: MastraOnFinishCallback; onError?: ({ error }: { error: Error | string; }) => void | Promise; onSuspended?: (data: AgentSuspendedEventData) => void | Promise; }): Promise, 'runId'> & { runId: string; }>; /** * Get the workflow instance for direct execution. * Lazily creates the workflow and registers Mastra on it (needed for * getAgentById in execution steps). */ getWorkflow(): import("../../workflows").Workflow[], "durable-agentic-loop", unknown, { __workflowKind: "durable-agent"; runId: string; agentId: string; messageListState: any; toolsMetadata: any[]; modelConfig: { provider: string; modelId: string; specificationVersion?: string | undefined; settings?: Record | undefined; providerOptions?: Record | undefined; }; options: any; state: any; messageId: string; agentName?: string | undefined; modelList?: { id: string; config: { provider: string; modelId: string; specificationVersion?: string | undefined; originalConfig?: string | Record | undefined; providerOptions?: Record | undefined; }; maxRetries: number; enabled: boolean; }[] | undefined; agentSpanData?: any; modelSpanData?: any; requestContextEntries?: Record | undefined; }, { messageListState: any; messageId: string; stepResult: any; output: { usage: any; steps: any[]; text?: string | undefined; }; state: any; }, { messageListState: any; messageId: string; stepResult: any; output: { usage: any; steps: any[]; text?: string | undefined; }; state: any; }, unknown>; /** * @deprecated Use `stream(messages, { untilIdle: true })` instead. * * Stream until all background tasks complete and the agent is idle. * Mirrors the regular Agent's streamUntilIdle but adapted for durable execution. */ streamUntilIdle(messages: MessageListInput, streamOptions?: DurableAgentStreamOptions & { maxIdleMs?: number; }): Promise>; /** * Prepare for durable execution without starting it. */ prepare(messages: MessageListInput, options?: AgentExecutionOptions): Promise<{ runId: string; messageId: string; workflowInput: DurableAgenticWorkflowInput; registryEntry: import("./types").RunRegistryEntry; threadId: string | undefined; resourceId: string | undefined; }>; /** * Get the durable workflows required by this agent. * Called by Mastra during agent registration. * @internal */ getDurableWorkflows(): import("../../workflows").Workflow[], "durable-agentic-loop", unknown, { __workflowKind: "durable-agent"; runId: string; agentId: string; messageListState: any; toolsMetadata: any[]; modelConfig: { provider: string; modelId: string; specificationVersion?: string | undefined; settings?: Record | undefined; providerOptions?: Record | undefined; }; options: any; state: any; messageId: string; agentName?: string | undefined; modelList?: { id: string; config: { provider: string; modelId: string; specificationVersion?: string | undefined; originalConfig?: string | Record | undefined; providerOptions?: Record | undefined; }; maxRetries: number; enabled: boolean; }[] | undefined; agentSpanData?: any; modelSpanData?: any; requestContextEntries?: Record | undefined; }, { messageListState: any; messageId: string; stepResult: any; output: { usage: any; steps: any[]; text?: string | undefined; }; state: any; }, { messageListState: any; messageId: string; stepResult: any; output: { usage: any; steps: any[]; text?: string | undefined; }; state: any; }, unknown>[]; /** * Set the Mastra instance. * Called by the durable agent registration path in addAgent(). * Delegates to __registerMastra so the pubsub wiring and agent * registration happen regardless of which entry point is called first. * @internal */ __setMastra(mastra: Mastra): void; /** * Register the Mastra instance. * Called by Mastra during agent registration (normal Agent path). * * Also wires mastra.pubsub as the inner pubsub (if the user didn't provide * a custom one), so that the OBSERVE_AGENT_STREAM_ROUTE handler can subscribe * to the same PubSub instance that this agent publishes to. * @internal */ __registerMastra(mastra: Mastra): void; } //# sourceMappingURL=durable-agent.d.ts.map