fix(core): remove saveStepMessages and disable savePerStep when observational memory is enabled (#15652)

Co-authored-by: Mastra Code (opencode/kimi-k2.6) <noreply@mastra.ai>

Author: TylerBarnes

.changeset/puny-actors-vanish.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fixed interaction between savePerStep and observational memory that caused message duplication. The saveStepMessages method redundantly re-added response messages to the message list on every step, duplicating them. Additionally, savePerStep is now force-disabled when observational memory is enabled, since OM handles its own per-step persistence and the two features conflict.

docs/src/content/en/reference/agents/generate.mdx

@@ -766,7 +766,8 @@ const response = await agent.generate('Help me organize my day', {
             name: 'savePerStep',
             type: 'boolean',
             isOptional: true,
-            description: 'Save messages incrementally after each generation step completes (default: false).',
+            description:
+              'Save messages incrementally after each generation step completes (default: false). Disabled internally when observational memory is enabled.',
           },
           {
             name: 'providerOptions',

docs/src/content/en/reference/agents/generateLegacy.mdx

@@ -336,7 +336,8 @@ await agent.generateLegacy('message for agent')
             name: 'savePerStep',
             type: 'boolean',
             isOptional: true,
-            description: 'Save messages incrementally after each stream step completes (default: false).',
+            description:
+              'Save messages incrementally after each generation step completes (default: false). Disabled internally when observational memory is enabled.',
           },
           {
             name: 'providerOptions',

packages/core/src/agent/agent-legacy.ts

@@ -165,8 +165,6 @@ export interface AgentLegacyCapabilities {
     instructions?: DynamicArgument<string>;
     minMessages?: number;
   };
-  /** Save step messages */
-  saveStepMessages(args: { result: any; messageList: MessageList; runId: string }): Promise<void>;
   /** Convert instructions to string */
   convertInstructionsToString(instructions: AgentInstructions): string;
   /** Options for tracing policy */
@@ -834,12 +832,6 @@ export class AgentLegacyHandler {
                 });
                 threadCreatedByStep = true;
               }
-
-              await this.capabilities.saveStepMessages({
-                result: props,
-                messageList,
-                runId,
-              });
             }
 
             return onStepFinish?.({ ...props, runId });

packages/core/src/agent/agent.ts

@@ -1412,7 +1412,6 @@ export class Agent<
         getMostRecentUserMessage: this.getMostRecentUserMessage.bind(this),
         genTitle: this.genTitle.bind(this),
         resolveTitleGenerationConfig: this.resolveTitleGenerationConfig.bind(this),
-        saveStepMessages: this.saveStepMessages.bind(this),
         convertInstructionsToString: this.#convertInstructionsToString.bind(this),
         tracingPolicy: this.#options?.tracingPolicy,
         resolvedVersionId: this.toRawConfig()?.resolvedVersionId as string | undefined,
@@ -4530,47 +4529,6 @@ export class Agent<
     return tools;
   }
 
-  /**
-   * Adds response messages from a step to the MessageList and schedules persistence.
-   * This is used for incremental saving: after each agent step, messages are added to a save queue
-   * and a debounced save operation is triggered to avoid redundant writes.
-   *
-   * @param result - The step result containing response messages.
-   * @param messageList - The MessageList instance for the current thread.
-   * @param threadId - The thread ID.
-   * @param memoryConfig - The memory configuration for saving.
-   * @param runId - (Optional) The run ID for logging.
-   * @internal
-   */
-  private async saveStepMessages({
-    result,
-    messageList,
-    runId,
-  }: {
-    result: any;
-    messageList: MessageList;
-    runId?: string;
-  }) {
-    try {
-      // Prefer dbMessages (MastraDBMessage[] with original IDs) over response.messages
-      // (ModelMessage[] without IDs) to avoid generating new IDs during format conversion
-      const stepResponseMessages = result.response.dbMessages?.length
-        ? result.response.dbMessages
-        : result.response.messages;
-      if (stepResponseMessages?.length) {
-        messageList.add(stepResponseMessages, 'response');
-      }
-      // Message saving is now handled by MessageHistory output processor
-    } catch (e) {
-      this.logger.error('Error adding messages on step finish', {
-        agent: this.name,
-        error: e,
-        runId,
-      });
-      throw e;
-    }
-  }
-
   async #runScorers({
     messageList,
     runId,
@@ -5013,7 +4971,6 @@ export class Agent<
         '_agentNetworkAppend' in this
           ? Boolean((this as unknown as { _agentNetworkAppend: unknown })._agentNetworkAppend)
           : undefined,
-      saveStepMessages: this.saveStepMessages.bind(this),
       convertTools: this.convertTools.bind(this),
       getMemoryMessages: this.getMemoryMessages.bind(this),
       runInputProcessors: this.__runInputProcessors.bind(this),

packages/core/src/agent/agent.types.ts

@@ -450,7 +450,7 @@ export type AgentExecutionOptionsBase<OUTPUT> = {
   /** Unique identifier for this execution run */
   runId?: string;
 
-  /** Save messages incrementally after each stream step completes (default: false). */
+  /** Save messages incrementally after each stream step completes (default: false). Is disabled internally when observational memory is enabled, as OM handles its own message saving */
   savePerStep?: boolean;
 
   /** Request Context containing dynamic configuration and state */

packages/core/src/agent/workflows/prepare-stream/map-results-step.ts

@@ -76,7 +76,9 @@ export function createMapResultsStep<OUTPUT = undefined>({
       requestContext,
       messageList: memoryData.messageList,
       onStepFinish: async (props: any) => {
-        if (options.savePerStep && !memoryConfig?.readOnly) {
+        // When OM is enabled saving per step corrupts things because OM handles its own saving
+        const shouldSavePerStep = options.savePerStep && !memoryConfig?.observationalMemory;
+        if (shouldSavePerStep && !memoryConfig?.readOnly) {
           if (!memoryData.threadExists && !threadCreatedByStep && memory && memoryData.thread) {
             await memory.createThread({
               threadId: memoryData.thread?.id,
@@ -89,12 +91,6 @@ export function createMapResultsStep<OUTPUT = undefined>({
             threadCreatedByStep = true;
           }
 
-          await capabilities.saveStepMessages({
-            result: props,
-            messageList: memoryData.messageList!,
-            runId,
-          });
-
           if (saveQueueManager && memoryData.thread?.id) {
             await saveQueueManager.flushMessages(memoryData.messageList!, memoryData.thread.id, memoryConfig);
           }

packages/core/src/agent/workflows/prepare-stream/schema.ts

@@ -22,7 +22,6 @@ export type AgentCapabilities = {
   generateMessageId: Mastra['generateId'];
   mastra?: Mastra;
   _agentNetworkAppend?: boolean;
-  saveStepMessages: Agent['saveStepMessages'];
   convertTools: Agent['convertTools'];
   runInputProcessors: Agent['__runInputProcessors'];
   executeOnFinish: (args: AgentExecuteOnFinishOptions) => Promise<void>;

packages/core/src/harness/harness.ts

@@ -1328,6 +1328,7 @@ export class Harness<TState = {}> {
         maxSteps: 1000,
         // Harness supports suspending + resuming streams (tool approvals, tool suspensions, workflows).
         // Persisting per-step snapshots ensures `resumeStream()` can load state reliably (especially in CI).
+        // Doesn't do anything when OM is enabled though, OM does its own saving per step
         savePerStep: true,
         requireToolApproval: !isYolo,
         modelSettings: { temperature: 1 },