fix: add processInputStep response id rotation (#13887)

This adds the ability for `processInputStep` to rotate the active
response message id, and wires the loop to honor that rotated id through
streamed output and persistence.

Before, the active response id was effectively fixed for the lifetime of
the response. After, a processor can explicitly request a new response
id during `processInputStep`, and the rest of the loop keeps using that
new id consistently.

```ts
// before
processInputStep({ messageId })
// response continues under the original id
```

```ts
// after
processInputStep({ messageId, rotateResponseMessageId })
// response continues under the rotated id
```

This also adds the small observational-memory compatibility guard needed
for that new core capability, so OM fails fast when paired with an older
`@mastra/core` that doesn't support response id rotation yet.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Response message ID rotation during processor execution and streaming.
  * New core feature flag: request-response-id-rotation.

* **Bug Fixes**
* Streamed assistant IDs and persisted memory now use the rotated
response ID.
* Avoid merging when a replacement target is sealed to preserve message
integrity.

* **Chores**
  * Observational Memory now fails fast if core lacks rotation support.

* **Tests**
* Added tests verifying rotation behavior across processors and
streaming.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Mastra Code (openai/gpt-5.3-codex) <noreply@mastra.ai>

Author: TylerBarnes

.changeset/strict-squids-tease.md

@@ -0,0 +1,5 @@
+---
+'@mastra/memory': patch
+---
+
+Added a compatibility guard so observational memory now fails fast when @mastra/core does not support request-response-id-rotation.

.changeset/vast-baboons-own.md

@@ -0,0 +1,7 @@
+---
+'@mastra/core': patch
+---
+
+Added processor-driven response message ID rotation so streamed assistant IDs use the rotated ID.
+
+Processors that run outside the agent loop no longer need synthetic response message IDs.

packages/core/src/agent/__tests__/stream-id.test.ts

@@ -6,6 +6,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { z } from 'zod';
 import { Mastra } from '../../mastra';
 import { MockMemory } from '../../memory/mock';
+import type { Processor } from '../../processors';
 import { createTool } from '../../tools';
 import { Agent } from '../agent';
 
@@ -279,6 +280,73 @@ describe('Stream ID Consistency', () => {
     expect(customIdGenerator).toHaveBeenCalled();
   });
 
+  it('should let processInputStep rotate the active response message ID for stream output and persistence', async () => {
+    let initialMessageId: string | undefined;
+    let rotatedMessageId: string | undefined;
+
+    const rotateMessageIdProcessor = {
+      id: 'rotate-response-message-id-processor',
+      processInputStep: async ({ messageId, rotateResponseMessageId }) => {
+        initialMessageId = messageId;
+        rotatedMessageId = rotateResponseMessageId?.();
+        return {};
+      },
+    } satisfies Processor;
+
+    const model = new MockLanguageModelV2({
+      doStream: async () => ({
+        rawCall: { rawPrompt: null, rawSettings: {} },
+        warnings: [],
+        stream: convertArrayToReadableStream([
+          { type: 'stream-start', warnings: [] },
+          { type: 'response-metadata', id: 'provider-msg-xyz123', modelId: 'mock-model-id', timestamp: new Date(0) },
+          { type: 'text-start', id: 'text-1' },
+          { type: 'text-delta', id: 'text-1', delta: 'rotated ' },
+          { type: 'text-delta', id: 'text-1', delta: 'response id' },
+          { type: 'text-end', id: 'text-1' },
+          {
+            type: 'finish',
+            finishReason: 'stop',
+            usage: { inputTokens: 10, outputTokens: 20, totalTokens: 30 },
+          },
+        ]),
+      }),
+    });
+
+    const agent = new Agent({
+      id: 'test-agent',
+      name: 'Test Agent V2 Rotated Response ID',
+      instructions: 'You are a helpful assistant.',
+      model,
+      memory,
+      inputProcessors: [rotateMessageIdProcessor],
+    });
+
+    agent.__registerMastra(mastra);
+
+    const threadId = randomUUID();
+    const resourceId = 'test-resource';
+    const stream = await agent.stream('Hello!', { memory: { thread: threadId, resource: resourceId } });
+
+    for await (const _chunk of stream.fullStream) {
+    }
+
+    const res = await stream.response;
+    const messageId = res?.uiMessages?.[0]?.id;
+
+    expect(initialMessageId).toBeDefined();
+    expect(rotatedMessageId).toBeDefined();
+    expect(rotatedMessageId).not.toBe(initialMessageId);
+    expect(messageId).toBe(rotatedMessageId);
+
+    const rotatedResult = await memory.recall({ threadId, perPage: 0, include: [{ id: rotatedMessageId! }] });
+    expect(rotatedResult.messages).toHaveLength(1);
+    expect(rotatedResult.messages[0].id).toBe(rotatedMessageId!);
+
+    const initialResult = await memory.recall({ threadId, perPage: 0, include: [{ id: initialMessageId! }] });
+    expect(initialResult.messages).toHaveLength(0);
+  });
+
   it('should return generate response IDs that match database-saved message IDs (V2 model)', async () => {
     const model = new MockLanguageModelV2({
       doGenerate: async () => ({
@@ -338,13 +406,11 @@ describe('Stream ID Consistency', () => {
     const responseMessageId = result.response?.uiMessages?.[0]?.id;
     expect(responseMessageId).toBeDefined();
 
-    // Verify the response message ID can be found in memory
     const recalled = await memory.recall({ threadId, include: [{ id: responseMessageId! }] });
     const messageById = recalled.messages.find(m => m.id === responseMessageId);
     expect(messageById).toBeDefined();
     expect(messageById!.id).toBe(responseMessageId);
 
-    // Verify no duplicate assistant messages exist (only 1 user + 1 assistant)
     const allMessages = await memory.recall({ threadId });
     const assistantMessages = allMessages.messages.filter(m => m.role === 'assistant');
     expect(assistantMessages).toHaveLength(1);
@@ -440,25 +506,21 @@ describe('Stream ID Consistency', () => {
       memory: { thread: threadId, resource: resourceId },
     });
 
-    // Consume the entire stream
     await streamResult.consumeStream();
 
     const response = await streamResult.response;
     const uiMessageIds = response?.uiMessages?.map(m => m.id) || [];
 
-    // Verify all response message IDs can be found in memory
     const allRecalled = await memory.recall({ threadId });
 
     for (const uiMsgId of uiMessageIds) {
       const matchInMemory = allRecalled.messages.find(m => m.id === uiMsgId);
       expect(matchInMemory, `uiMessage ID ${uiMsgId} should exist in memory`).toBeDefined();
     }
 
-    // Verify fullOutput.messages IDs match response uiMessage IDs
     const fullOutput = await streamResult.getFullOutput();
     const responseMessageIds = fullOutput.messages.filter(m => m.role === 'assistant').map(m => m.id);
 
-    // Every uiMessage ID should appear in fullOutput.messages
     for (const uiMsgId of uiMessageIds) {
       expect(responseMessageIds, `uiMessage ID ${uiMsgId} should appear in fullOutput.messages`).toContain(uiMsgId);
     }

packages/core/src/agent/message-list/message-list.ts

@@ -824,16 +824,17 @@ export class MessageList {
         }
       }
     }
-    // If the last message is an assistant message and the new message is also an assistant message, merge them together and update tool calls with results
-    // Use MessageMerger to handle the complex merge logic
+
+    const replacementTarget = exists && id ? this.messages.find(m => m.id === id) : undefined;
+    const hasSealedReplacementTarget = !!replacementTarget && MessageMerger.isSealed(replacementTarget);
+
+    // Keep this replacement-target guard here instead of MessageMerger.shouldMerge().
+    // shouldMerge() only decides whether to append to the latest assistant message,
+    // but replace-by-id can target an older sealed message elsewhere in the list.
     const isLatestFromMemory = latestMessage ? this.memoryMessages.has(latestMessage) : false;
-    const shouldMerge = MessageMerger.shouldMerge(
-      latestMessage,
-      messageV2,
-      messageSource,
-      isLatestFromMemory,
-      this._agentNetworkAppend,
-    );
+    const shouldMerge =
+      !hasSealedReplacementTarget &&
+      MessageMerger.shouldMerge(latestMessage, messageV2, messageSource, isLatestFromMemory, this._agentNetworkAppend);
 
     if (shouldMerge && latestMessage) {
       // Delegate merge logic to MessageMerger

packages/core/src/features/index.ts

@@ -14,4 +14,10 @@
  * ```
  */
 // Add feature flags here as new features are introduced
-export const coreFeatures = new Set<string>(['observationalMemory', 'asyncBuffering', 'workspaces-v1', 'datasets']);
+export const coreFeatures = new Set<string>([
+  'observationalMemory',
+  'asyncBuffering',
+  'request-response-id-rotation',
+  'workspaces-v1',
+  'datasets',
+]);

packages/core/src/loop/workflows/agentic-execution/llm-execution-step.ts

@@ -1,7 +1,7 @@
 import { ReadableStream } from 'node:stream/web';
 import { isAbortError } from '@ai-sdk/provider-utils-v5';
 import type { LanguageModelV2Usage } from '@ai-sdk/provider-v5';
-import { APICallError } from '@internal/ai-sdk-v5';
+import { APICallError, generateId } from '@internal/ai-sdk-v5';
 import type { CallSettings, ToolChoice, ToolSet } from '@internal/ai-sdk-v5';
 import type { StructuredOutputOptions } from '../../../agent';
 import type { MastraDBMessage, MessageList } from '../../../agent/message-list';
@@ -546,9 +546,9 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
     execute: async ({ inputData, bail, tracingContext }) => {
       currentIteration++;
 
-      const messageId = inputData.isTaskCompleteCheckFailed
+      let currentMessageId = inputData.isTaskCompleteCheckFailed
         ? `${messageIdPassed}-${currentIteration}`
-        : messageIdPassed;
+        : inputData.messageId || messageIdPassed;
       // Start the MODEL_STEP span at the beginning of LLM execution
       modelSpanTracker?.startStep();
 
@@ -583,6 +583,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
         }
 
         const currentStep: {
+          messageId: string;
           model: MastraLanguageModel;
           tools?: TOOLS | undefined;
           toolChoice?: ToolChoice<TOOLS> | undefined;
@@ -592,6 +593,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
           structuredOutput?: StructuredOutputOptions<OUTPUT>;
           workspace?: Workspace;
         } = {
+          messageId: currentMessageId,
           model,
           tools,
           toolChoice,
@@ -631,6 +633,12 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
               requestContext,
               model,
               steps: inputData.output?.steps || [],
+              messageId: currentStep.messageId,
+              rotateResponseMessageId: () => {
+                currentMessageId = _internal?.generateId?.() ?? generateId();
+                currentStep.messageId = currentMessageId;
+                return currentMessageId;
+              },
               tools,
               toolChoice,
               activeTools: activeTools as string[] | undefined,
@@ -679,7 +687,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
                     },
                   }),
                   messageList,
-                  messageId,
+                  messageId: currentStep.messageId,
                   options: { runId },
                 }),
                 runState,
@@ -817,7 +825,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
                     payload: {
                       request: request || {},
                       warnings: warnings || [],
-                      messageId: messageId,
+                      messageId: currentStep.messageId,
                     },
                   });
                 },
@@ -838,7 +846,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
           },
           stream: modelResult as ReadableStream<ChunkType<OUTPUT>>,
           messageList,
-          messageId,
+          messageId: currentStep.messageId,
           options: {
             runId,
             toolCallStreaming,
@@ -863,7 +871,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
             outputStream,
             includeRawChunks,
             tools: currentStep.tools,
-            messageId,
+            messageId: currentStep.messageId,
             messageList,
             runState,
             options,
@@ -965,7 +973,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
         const text = outputStream._getImmediateText();
 
         return bail({
-          messageId,
+          messageId: outputStream.messageId,
           stepResult: {
             reason: 'tripwire',
             warnings,
@@ -1012,7 +1020,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
 
       if (toolCalls.length > 0) {
         const message: MastraDBMessage = {
-          id: messageId,
+          id: outputStream.messageId,
           role: 'assistant' as const,
           content: {
             format: 2,
@@ -1166,7 +1174,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
       // Without this, the LLM sees the rejected assistant response in its prompt on retry,
       // which confuses models and often causes empty text responses.
       if (shouldRetry) {
-        messageList.removeByIds([messageId]);
+        messageList.removeByIds([outputStream.messageId]);
       }
 
       // Build retry feedback text if retrying
@@ -1195,7 +1203,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
       const nextProcessorRetryCount = shouldRetry ? currentProcessorRetryCount + 1 : currentProcessorRetryCount;
 
       return {
-        messageId,
+        messageId: outputStream.messageId,
         stepResult: {
           reason: stepReason,
           warnings,

packages/core/src/processors/index.ts

@@ -119,6 +119,10 @@ export interface ProcessInputStepArgs<TTripwireMetadata = unknown> extends Proce
   /** The current step number (0-indexed) */
   stepNumber: number;
   steps: Array<StepResult<any>>;
+  /** The active assistant response message ID for this step, when this processor is running inside an agent loop */
+  messageId?: string;
+  /** Mark the current assistant response message ID as complete and rotate to a fresh one, when supported by the caller */
+  rotateResponseMessageId?: () => string;
 
   /** All system messages (agent instructions, user-provided, memory) for read/modify access */
   systemMessages: CoreMessageV4[];
@@ -149,7 +153,14 @@ export interface ProcessInputStepArgs<TTripwireMetadata = unknown> extends Proce
   retryCount: number;
 }
 
-export type RunProcessInputStepArgs = Omit<ProcessInputStepArgs, 'messages' | 'systemMessages' | 'abort' | 'state'>;
+export type RunProcessInputStepArgs = Omit<
+  ProcessInputStepArgs,
+  'messages' | 'systemMessages' | 'abort' | 'state' | 'messageId' | 'rotateResponseMessageId' | 'retryCount'
+> & {
+  messageId?: string;
+  rotateResponseMessageId?: () => string;
+  retryCount?: number;
+};
 
 /**
  * Result from processInputStep method
@@ -159,6 +170,8 @@ export type RunProcessInputStepArgs = Omit<ProcessInputStepArgs, 'messages' | 's
  */
 export type ProcessInputStepResult = {
   model?: LanguageModelV2 | ModelRouterModelId | OpenAICompatibleConfig | MastraLanguageModel;
+  /** Override the active assistant response message ID for this step */
+  messageId?: string;
   /** Replace tools for this step - accepts both AI SDK tools and Mastra createTool results */
   tools?: Record<string, unknown>;
   toolChoice?: ToolChoice<any>;

packages/core/src/processors/process-input-step.test.ts

@@ -139,6 +139,55 @@ describe('processInputStep', () => {
       expect(typeof runner.runProcessInputStep).toBe('function');
     });
 
+    it('should rotate the active response message id for later processors and the final step result', async () => {
+      const seenMessageIds: string[] = [];
+      const rotateResponseMessageId = vi.fn(() => 'response-2');
+
+      const rotateProcessor: Processor = {
+        id: 'rotate-processor',
+        processInputStep: async ({ messageId, rotateResponseMessageId }) => {
+          if (messageId) {
+            seenMessageIds.push(messageId);
+          }
+          rotateResponseMessageId?.();
+          return {};
+        },
+      };
+
+      const observeProcessor: Processor = {
+        id: 'observe-processor',
+        processInputStep: async ({ messageId }) => {
+          if (messageId) {
+            seenMessageIds.push(messageId);
+          }
+          return {};
+        },
+      };
+
+      const runner = new ProcessorRunner({
+        inputProcessors: [rotateProcessor, observeProcessor],
+        outputProcessors: [],
+        logger: mockLogger,
+        agentName: 'test-agent',
+      });
+
+      const messageList = new MessageList({ threadId: 'test-thread' });
+      messageList.add([createMessage('Hello')], 'input');
+
+      const result = await runner.runProcessInputStep({
+        messageList,
+        stepNumber: 0,
+        model: createMockModel(),
+        steps: [],
+        messageId: 'response-1',
+        rotateResponseMessageId,
+      });
+
+      expect(rotateResponseMessageId).toHaveBeenCalledTimes(1);
+      expect(seenMessageIds).toEqual(['response-1', 'response-2']);
+      expect(result.messageId).toBe('response-2');
+    });
+
     it('should be callable at each step with growing message history', async () => {
       let processInputStepCallCount = 0;
       const stepNumbers: number[] = [];