From a127db6557c30d1003c7732caa59d36aa0f16e7c Mon Sep 17 00:00:00 2001 From: shamAnimates <145093437+shamAnimates@users.noreply.github.com> Date: Sun, 16 Aug 2026 16:19:54 +0530 Subject: [PATCH] fix(tools): scope Mastra memory cache per request --- packages/tools/src/mastra/processor.ts | 34 +++++- .../tools/test/mastra/integration.test.ts | 5 +- packages/tools/test/mastra/unit.test.ts | 111 +++++++++++++++++- 3 files changed, 140 insertions(+), 10 deletions(-) diff --git a/packages/tools/src/mastra/processor.ts b/packages/tools/src/mastra/processor.ts index 4998c150..64d35571 100644 --- a/packages/tools/src/mastra/processor.ts +++ b/packages/tools/src/mastra/processor.ts @@ -50,7 +50,29 @@ interface ProcessorContext { addMemory: "always" | "never" logger: Logger promptTemplate?: PromptTemplate - memoryCache: MemoryCache +} + +const REQUEST_MEMORY_CACHE_KEY = "supermemory.memoryCache" + +/** + * Gets the cache scoped to the current Mastra request. + * + * Mastra processor instances can be reused across requests, while `state` is + * created for one request and shared across that request's processor calls. + */ +function getRequestMemoryCache( + state?: Record, +): MemoryCache { + const existingCache = state?.[REQUEST_MEMORY_CACHE_KEY] + if (existingCache instanceof MemoryCache) { + return existingCache + } + + const memoryCache = new MemoryCache() + if (state) { + state[REQUEST_MEMORY_CACHE_KEY] = memoryCache + } + return memoryCache } /** @@ -72,7 +94,6 @@ function createProcessorContext( addMemory: options.addMemory ?? "always", logger, promptTemplate: options.promptTemplate, - memoryCache: new MemoryCache(), } } @@ -136,7 +157,7 @@ export class SupermemoryInputProcessor implements Processor { } async processInput(args: ProcessInputArgs): Promise { - const { messages, messageList, requestContext } = args + const { messages, messageList, requestContext, state } = args try { const queryText = extractQueryText( @@ -159,9 +180,10 @@ export class SupermemoryInputProcessor implements Processor { this.ctx.mode, queryText || "", ) + const memoryCache = getRequestMemoryCache(state) - if (this.ctx.memoryCache.has(turnKey)) { - const cachedMemories = this.ctx.memoryCache.get(turnKey) ?? "" + if (memoryCache.has(turnKey)) { + const cachedMemories = memoryCache.get(turnKey) ?? "" this.ctx.logger.debug("Using cached memories", { turnKey }) if (cachedMemories) { messageList.addSystem(cachedMemories, "supermemory") @@ -185,7 +207,7 @@ export class SupermemoryInputProcessor implements Processor { promptTemplate: this.ctx.promptTemplate, }) - this.ctx.memoryCache.set(turnKey, memories) + memoryCache.set(turnKey, memories) if (memories) { messageList.addSystem(memories, "supermemory") this.ctx.logger.debug("Injected memories into system prompt", { diff --git a/packages/tools/test/mastra/integration.test.ts b/packages/tools/test/mastra/integration.test.ts index 58b79a61..92382287 100644 --- a/packages/tools/test/mastra/integration.test.ts +++ b/packages/tools/test/mastra/integration.test.ts @@ -212,7 +212,7 @@ describe.skipIf(!shouldRunIntegration)( fetchSpy.mockRestore() }) - it("should cache memories for repeated calls with same message", async () => { + it("should cache memories for repeated calls within a request", async () => { const fetchSpy = vi.spyOn(globalThis, "fetch") const processor = new SupermemoryInputProcessor({ @@ -226,11 +226,13 @@ describe.skipIf(!shouldRunIntegration)( const messages: MastraDBMessage[] = [ createMessage("user", "Cache test message"), ] + const state: Record = {} const args1: ProcessInputArgs = { messages, systemMessages: [], messageList: createIntegrationMessageList(), + state, abort: vi.fn() as never, retryCount: 0, } @@ -245,6 +247,7 @@ describe.skipIf(!shouldRunIntegration)( messages, systemMessages: [], messageList: createIntegrationMessageList(), + state, abort: vi.fn() as never, retryCount: 0, } diff --git a/packages/tools/test/mastra/unit.test.ts b/packages/tools/test/mastra/unit.test.ts index f8ea2ceb..c3a51bb2 100644 --- a/packages/tools/test/mastra/unit.test.ts +++ b/packages/tools/test/mastra/unit.test.ts @@ -216,6 +216,7 @@ describe("SupermemoryInputProcessor", () => { const messages: MastraDBMessage[] = [createMessage("user", "Hello")] const firstMessageList = createMockMessageList() const secondMessageList = createMockMessageList() + const state: Record = {} await processor.processInput({ messages, @@ -223,7 +224,7 @@ describe("SupermemoryInputProcessor", () => { messageList: firstMessageList, abort: vi.fn() as never, retryCount: 0, - state: {}, + state, }) await processor.processInput({ messages, @@ -231,7 +232,7 @@ describe("SupermemoryInputProcessor", () => { messageList: secondMessageList, abort: vi.fn() as never, retryCount: 0, - state: {}, + state, }) expect(firstMessageList.addSystem).not.toHaveBeenCalled() @@ -239,7 +240,7 @@ describe("SupermemoryInputProcessor", () => { expect(fetchMock).toHaveBeenCalledTimes(1) }) - it("should use cached memories on second call with same message", async () => { + it("should use cached memories within the same request", async () => { fetchMock.mockResolvedValue({ ok: true, json: () => @@ -254,11 +255,13 @@ describe("SupermemoryInputProcessor", () => { }) const messages: MastraDBMessage[] = [createMessage("user", "Hello")] + const state: Record = {} const args1: ProcessInputArgs = { messages, systemMessages: [], messageList: createMockMessageList(), + state, abort: vi.fn() as never, retryCount: 0, } @@ -270,6 +273,7 @@ describe("SupermemoryInputProcessor", () => { messages, systemMessages: [], messageList: createMockMessageList(), + state, abort: vi.fn() as never, retryCount: 0, } @@ -278,6 +282,104 @@ describe("SupermemoryInputProcessor", () => { expect(fetchMock).toHaveBeenCalledTimes(1) }) + it("should not reuse an empty profile result across requests", async () => { + let callCount = 0 + fetchMock.mockImplementation(() => { + const currentCall = ++callCount + return Promise.resolve({ + ok: true, + json: () => + Promise.resolve( + currentCall === 1 + ? createMockProfileResponse() + : createMockProfileResponse(["Profile from second request"]), + ), + }) + }) + + const processor = new SupermemoryInputProcessor({ + containerTag: TEST_CONFIG.containerTag, + customId: TEST_CONFIG.customId, + apiKey: TEST_CONFIG.apiKey, + mode: "profile", + }) + const messages: MastraDBMessage[] = [createMessage("user", "Hello")] + const firstMessageList = createMockMessageList() + const secondMessageList = createMockMessageList() + + await processor.processInput({ + messages, + systemMessages: [], + messageList: firstMessageList, + state: {}, + abort: vi.fn() as never, + retryCount: 0, + }) + await processor.processInput({ + messages, + systemMessages: [], + messageList: secondMessageList, + state: {}, + abort: vi.fn() as never, + retryCount: 0, + }) + + expect(fetchMock).toHaveBeenCalledTimes(2) + expect(firstMessageList.addSystem).not.toHaveBeenCalled() + expect(secondMessageList.calls[0]?.args[0]).toContain( + "Profile from second request", + ) + }) + + it("should not share a cache when request state is unavailable", async () => { + let callCount = 0 + fetchMock.mockImplementation(() => { + const currentCall = ++callCount + return Promise.resolve({ + ok: true, + json: () => + Promise.resolve( + createMockProfileResponse([`Profile from call ${currentCall}`]), + ), + }) + }) + + const processor = new SupermemoryInputProcessor({ + containerTag: TEST_CONFIG.containerTag, + customId: TEST_CONFIG.customId, + apiKey: TEST_CONFIG.apiKey, + mode: "profile", + }) + const messages: MastraDBMessage[] = [createMessage("user", "Hello")] + const firstMessageList = createMockMessageList() + const secondMessageList = createMockMessageList() + + await processor.processInput({ + messages, + systemMessages: [], + messageList: firstMessageList, + state: undefined as never, + abort: vi.fn() as never, + retryCount: 0, + }) + await processor.processInput({ + messages, + systemMessages: [], + messageList: secondMessageList, + state: undefined as never, + abort: vi.fn() as never, + retryCount: 0, + }) + + expect(fetchMock).toHaveBeenCalledTimes(2) + expect(firstMessageList.calls[0]?.args[0]).toContain( + "Profile from call 1", + ) + expect(secondMessageList.calls[0]?.args[0]).toContain( + "Profile from call 2", + ) + }) + it("should refetch memories for different user message", async () => { let callCount = 0 fetchMock.mockImplementation(() => { @@ -297,11 +399,13 @@ describe("SupermemoryInputProcessor", () => { apiKey: TEST_CONFIG.apiKey, mode: "query", }) + const state: Record = {} const args1: ProcessInputArgs = { messages: [createMessage("user", "First message")], systemMessages: [], messageList: createMockMessageList(), + state, abort: vi.fn() as never, retryCount: 0, } @@ -313,6 +417,7 @@ describe("SupermemoryInputProcessor", () => { messages: [createMessage("user", "Different message")], systemMessages: [], messageList: createMockMessageList(), + state, abort: vi.fn() as never, retryCount: 0, }