fix: call processInputStep in legacy path for Observational Memory with v4 models (#13358)

## Summary
- Observational Memory (OM) was completely broken when using AI SDK v4
models (e.g., `openai('gpt-4o-mini')`) — the agent had no conversation
history and no observations
- The legacy stream/generate path (`agent-legacy.ts`) only ran the
`processInput` phase but never `processInputStep`. Since OM implements
`processInputStep` (not `processInput`), and `MessageHistory` is skipped
when OM is enabled, nothing loaded conversation context
- Adds `__runProcessInputStep()` to the Agent class and exposes it via
legacy capabilities so both `before()` paths call it after
`__runInputProcessors`

## Test plan
- [x] `pnpm build:core` passes
- [x] `pnpm build:memory` passes
- [ ] Start example agent server with `OM_DEBUG=1`, send messages to
script-agent, verify `[OM:processInputStep:ENTER]` appears in
om-debug.log
- [ ] Verify agent remembers context across messages when using v4
models with OM enabled

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

## Summary by CodeRabbit

## Bug Fixes
* Fixed Observational Memory not functioning with AI SDK v4 models in
the legacy execution path, restoring the ability to properly inject
conversation history and observations during agent processing.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: rphansen91

.changeset/icy-symbols-count.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fixed Observational Memory not working with AI SDK v4 models (legacy path). The legacy stream/generate path now calls processInputStep, enabling processors like Observational Memory to inject conversation history and observations.

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

@@ -113,6 +113,21 @@ export interface AgentLegacyCapabilities {
       processorId?: string;
     };
   }>;
+  /** Run processInputStep phase on input processors (for legacy path compatibility) */
+  __runProcessInputStep(args: {
+    requestContext: RequestContext;
+    tracingContext: TracingContext;
+    messageList: MessageList;
+    stepNumber?: number;
+  }): Promise<{
+    messageList: MessageList;
+    tripwire?: {
+      reason: string;
+      retry?: boolean;
+      metadata?: unknown;
+      processorId?: string;
+    };
+  }>;
   /** Get most recent user message */
   getMostRecentUserMessage(
     messages: Array<UIMessage | UIMessageWithMetadata>,
@@ -306,6 +321,26 @@ export class AgentLegacyHandler {
             tracingContext: innerTracingContext,
             messageList,
           });
+          // Run processInputStep for step 0 (legacy path compatibility)
+          if (!tripwire) {
+            const inputStepResult = await this.capabilities.__runProcessInputStep({
+              requestContext,
+              tracingContext: innerTracingContext,
+              messageList,
+              stepNumber: 0,
+            });
+            if (inputStepResult.tripwire) {
+              return {
+                messageObjects: [],
+                convertedTools,
+                threadExists: false,
+                thread: undefined,
+                messageList,
+                agentSpan,
+                tripwire: inputStepResult.tripwire,
+              };
+            }
+          }
           return {
             messageObjects: tripwire ? [] : messageList.get.all.prompt(),
             convertedTools,
@@ -391,8 +426,32 @@ export class AgentLegacyHandler {
         });
         messageList = processedMessageList;
 
-        // Messages are already processed by __runInputProcessors above
-        // which includes memory processors (WorkingMemory, MessageHistory, etc.)
+        // Run processInputStep phase for step 0 (legacy path compatibility).
+        // The v5 agentic loop runs this per-step in llm-execution-step, but the legacy
+        // path doesn't have that loop. This is needed for processors like Observational Memory
+        // that implement processInputStep (not processInput) to inject context.
+        if (!tripwire) {
+          const inputStepResult = await this.capabilities.__runProcessInputStep({
+            requestContext,
+            tracingContext: innerTracingContext,
+            messageList,
+            stepNumber: 0,
+          });
+          if (inputStepResult.tripwire) {
+            return {
+              convertedTools,
+              thread: threadObject,
+              messageList,
+              messageObjects: [],
+              agentSpan,
+              tripwire: inputStepResult.tripwire,
+              threadExists: !!existingThread,
+            };
+          }
+        }
+
+        // Messages are already processed by __runInputProcessors and __runProcessInputStep above
+        // which includes memory processors (WorkingMemory, MessageHistory, OM, etc.)
         const processedList = messageList.get.all.prompt();
 
         return {

packages/core/src/agent/agent.ts

@@ -1128,6 +1128,7 @@ export class Agent<
         convertTools: this.convertTools.bind(this),
         getMemoryMessages: (...args) => this.getMemoryMessages(...args),
         __runInputProcessors: this.__runInputProcessors.bind(this),
+        __runProcessInputStep: this.__runProcessInputStep.bind(this),
         getMostRecentUserMessage: this.getMostRecentUserMessage.bind(this),
         genTitle: this.genTitle.bind(this),
         resolveTitleGenerationConfig: this.resolveTitleGenerationConfig.bind(this),
@@ -2012,6 +2013,82 @@ export class Agent<
     };
   }
 
+  /**
+   * Runs processInputStep phase on input processors.
+   * Used by legacy path to execute per-step input processing (e.g., Observational Memory)
+   * that would otherwise only run in the v5 agentic loop.
+   * @internal
+   */
+  private async __runProcessInputStep({
+    requestContext,
+    tracingContext,
+    messageList,
+    stepNumber = 0,
+    processorStates,
+  }: {
+    requestContext: RequestContext;
+    tracingContext: TracingContext;
+    messageList: MessageList;
+    stepNumber?: number;
+    processorStates?: Map<string, ProcessorState>;
+  }): Promise<{
+    messageList: MessageList;
+    tripwire?: {
+      reason: string;
+      retry?: boolean;
+      metadata?: unknown;
+      processorId?: string;
+    };
+  }> {
+    let tripwire: { reason: string; retry?: boolean; metadata?: unknown; processorId?: string } | undefined;
+
+    if (this.#inputProcessors || this.#memory) {
+      const runner = await this.getProcessorRunner({
+        requestContext,
+        processorStates,
+      });
+      try {
+        const llm = await this.getLLM({ requestContext });
+        const model = llm.getModel();
+        await runner.runProcessInputStep({
+          messageList,
+          stepNumber,
+          steps: [],
+          tracingContext,
+          requestContext,
+          // Cast needed: legacy v1 models return LanguageModelV1 which doesn't satisfy MastraLanguageModel.
+          // OM's processInputStep doesn't use the model parameter, so this is safe.
+          model: model as MastraLanguageModel,
+          retryCount: 0,
+        });
+      } catch (error) {
+        if (error instanceof TripWire) {
+          tripwire = {
+            reason: error.message,
+            retry: error.options?.retry,
+            metadata: error.options?.metadata,
+            processorId: error.processorId,
+          };
+        } else {
+          throw new MastraError(
+            {
+              id: 'AGENT_INPUT_STEP_PROCESSOR_ERROR',
+              domain: ErrorDomain.AGENT,
+              category: ErrorCategory.USER,
+              text: `[Agent:${this.name}] - Input step processor error`,
+            },
+            error,
+          );
+        }
+      }
+    }
+
+    return {
+      messageList,
+      tripwire,
+    };
+  }
+
   /**
    * Executes output processors on the message list after LLM processing.
    * @internal

packages/memory/src/processors/observational-memory/observational-memory.ts

@@ -3325,8 +3325,13 @@ ${suggestedResponse}
     const { messageList, requestContext, stepNumber, state: _state, writer, abortSignal, abort } = args;
     const state = _state ?? ({} as Record<string, unknown>);
 
+    omDebug(
+      `[OM:processInputStep:ENTER] step=${stepNumber}, hasMastraMemory=${!!requestContext?.get('MastraMemory')}, hasMemoryInfo=${!!messageList?.serialize()?.memoryInfo?.threadId}`,
+    );
+
     const context = this.getThreadContext(requestContext, messageList);
     if (!context) {
+      omDebug(`[OM:processInputStep:NO-CONTEXT] getThreadContext returned null — returning early`);
       return messageList;
     }