fix(core): restore state signals without thread scans (#18182)

This fixes long-thread state signal restore by loading tracked signal
messages directly by id instead of scanning thread history.

Before:
```ts
await memoryStore.listMessages({
  threadId,
  resourceId,
  perPage: false,
});
```

After:
```ts
await memoryStore.listMessagesById({ messageIds: trackedSignalIds });
```

If tracked ids or id-based lookup are unavailable, we now fall back to
local in-memory state instead of doing an unbounded thread scan.

Verified with:
`pnpm --filter ./packages/core test -- src/processors/runner.test.ts
--bail 1 --reporter=dot`
`pnpm --filter ./packages/core check`


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

## ELI5

Imagine you have a very long list of messages and need to find a few
specific ones. The old way was to look through every single message in
the list one by one until you found what you needed. The new way is much
smarter: it says "give me these specific messages by their ID" instead,
skipping all the others. This makes it much faster when threads get
really long.

## Overview

This PR improves the performance of state signal restoration in
long-running threads by eliminating expensive full-thread message scans.
Instead of iterating through all messages in a thread's history, the
system now uses a targeted ID-based lookup to fetch only the tracked
signal messages it needs.

## Changes

**Core Logic Update** (`packages/core/src/agent/state-signals.ts`)
- Modified `resolveStateSignalHistory` to collect tracked signal message
IDs from `tracking.activeCopies` and `tracking.lastSnapshotSignalId`
upfront
- Replaced `memoryStore.listMessages({ threadId, resourceId, perPage:
false })` (full thread scan) with `memoryStore.listMessagesById({
messageIds })` (targeted lookup)
- Removed the `resourceId` parameter from the function signature since
it's no longer needed for the more efficient ID-based lookup
- Graceful degradation: if `listMessagesById` is unavailable or tracking
yields no IDs, the function returns the local in-memory state without
attempting expensive fallbacks

**Caller Update** (`packages/core/src/processors/runner.ts`)
- Removed the now-unnecessary `resourceId` argument from the
`resolveStateSignalHistory` call

**Test Updates** (`packages/core/src/processors/runner.test.ts`)
- Updated mock to provide both `listMessages` and `listMessagesById`
methods
- Changed test expectations to verify that only `listMessagesById` is
called (with the specific message IDs) and `listMessages` is not called

## Impact

This optimization prevents performance degradation as thread histories
grow longer, especially in long-running conversation scenarios where
state signals need to be restored. The changes were verified against
existing test suites in the core package.

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

---------

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

Author: TylerBarnes

.changeset/cuddly-ears-grab.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Improved state signal restoration so long threads can resume without scanning every message.

packages/core/src/agent/state-signals.ts

@@ -150,14 +150,12 @@ export async function resolveStateSignalHistory({
   messageList,
   memory,
   threadId,
-  resourceId,
   stateId,
   tracking,
 }: {
   messageList: MessageList;
   memory: MastraMemory;
   threadId: string;
-  resourceId: string;
   stateId: string;
   tracking?: StateSignalTracking;
 }): Promise<StateSignalHistory> {
@@ -172,12 +170,17 @@ export async function resolveStateSignalHistory({
   const memoryStore = await memory.storage.getStore('memory');
   if (!memoryStore) return { ...localHistory, contextWindow };
 
-  const storedMessages = await memoryStore.listMessages({
-    threadId,
-    resourceId,
-    perPage: false,
-    orderBy: { field: 'createdAt', direction: 'ASC' },
-  });
+  const trackedSignalIds = new Set<string>();
+  for (const activeCopy of tracking.activeCopies ?? []) {
+    trackedSignalIds.add(activeCopy.id);
+  }
+  trackedSignalIds.add(tracking.lastSnapshotSignalId);
+
+  if (trackedSignalIds.size === 0 || typeof memoryStore.listMessagesById !== 'function') {
+    return { ...localHistory, contextWindow };
+  }
+
+  const storedMessages = await memoryStore.listMessagesById({ messageIds: [...trackedSignalIds] });
   const storedStateSignals = dbMessagesToStateSignals(storedMessages.messages, stateId, threadId);
   const resolvedStateSignals = mergeStateSignals(storedStateSignals, localStateSignals);
 

packages/core/src/processors/runner.test.ts

@@ -3270,10 +3270,11 @@ describe('ProcessorRunner', () => {
         perPage: false,
         hasMore: false,
       }));
+      const listMessagesById = vi.fn(async () => ({ messages: storedMessages.get.all.db() }));
       const memory = {
         getThreadById: vi.fn(async () => thread),
         saveThread: vi.fn(async ({ thread }) => thread),
-        storage: { getStore: vi.fn(async () => ({ listMessages })) },
+        storage: { getStore: vi.fn(async () => ({ listMessages, listMessagesById })) },
       };
       const computeStateSignal = vi.fn((_args: any) => undefined);
 
@@ -3295,9 +3296,8 @@ describe('ProcessorRunner', () => {
         memory: memory as any,
       });
 
-      expect(listMessages).toHaveBeenCalledWith(
-        expect.objectContaining({ threadId: 'thread-1', resourceId: 'resource-1', perPage: false }),
-      );
+      expect(listMessages).not.toHaveBeenCalled();
+      expect(listMessagesById).toHaveBeenCalledWith({ messageIds: ['stored-snapshot'] });
       const computeArgs = computeStateSignal.mock.calls[0]?.[0];
       expect(computeArgs.activeStateSignals.map(signal => signal.id)).toEqual(
         expect.arrayContaining([snapshot.id, delta.id, localDelta.id]),

packages/core/src/processors/runner.ts

@@ -374,7 +374,6 @@ export class ProcessorRunner {
       messageList,
       memory: resolvedMemory,
       threadId: resolvedThreadId,
-      resourceId: resolvedResourceId,
       stateId,
       tracking,
     });