From 7cf076365dd4bcc1521059c377bb788c49f1578e Mon Sep 17 00:00:00 2001 From: abhay-codes07 Date: Mon, 13 Jul 2026 04:12:09 +0530 Subject: [PATCH] fix(pipecat): stop re-storing injected memory blocks as new memories MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LLMContext.get_messages() returns the context's live message list — the enhancement code depends on that, mutating it in place to inject the block. But the frame handler snapshotted that same reference *before* enhancement, so by the time it sliced out storable_messages the list already contained this turn's injected user-message, and _store_messages wrote the recalled memories back to Supermemory as a brand-new memory. Next turn recalls the stored block and re-stores it again, compounding every turn. (Previously masked by the dedup crash fixed in #1268 — with search results empty and profile facts present, the loop is live.) Two defenses: - copy the message list before enhancement mutates it, so the current turn's injection never enters the storage slice - filter any message carrying the memory tag out of storable_messages, which also covers the previous turn's injected block still sitting in the context (it is only replaced during the next enhancement, after the snapshot is taken) Regression tests drive process_frame with a live-list fake context and a mocked client: the injected block lands in the context but never in the storage payload, across both the same-turn and next-turn shapes. --- .../src/supermemory_pipecat/service.py | 21 +- .../tests/test_no_memory_restorage.py | 225 ++++++++++++++++++ 2 files changed, 243 insertions(+), 3 deletions(-) create mode 100644 packages/pipecat-sdk-python/tests/test_no_memory_restorage.py diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py index eb9d5fb66..58d2fc7b0 100644 --- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py +++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py @@ -34,6 +34,12 @@ MEMORY_TAG_PATTERN = re.compile(r".*?", re.DOTALL) +def _is_memory_injection(msg: Dict[str, Any]) -> bool: + """True if the message is a memory block this service injected.""" + content = msg.get("content") + return isinstance(content, str) and MEMORY_TAG_START in content + + class SupermemoryPipecatService(FrameProcessor): """Memory service that integrates Supermemory with Pipecat pipelines. @@ -292,7 +298,10 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: if context: try: - context_messages = context.get_messages() + # Snapshot: get_messages() returns the context's live list + # (enhancement below mutates it in place), so copy before + # this turn's memory block is injected. + context_messages = list(context.get_messages()) latest_user_message = get_last_user_message(context_messages) if latest_user_message: @@ -304,9 +313,15 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: except MemoryRetrievalError as e: logger.warning(f"Memory retrieval failed: {e}") - # Store unsent messages (user and assistant only) + # Store unsent messages (user and assistant only). Injected + # blocks are excluded — storing them would + # write recalled memories back to Supermemory as new + # memories, compounding every turn. storable_messages = [ - msg for msg in context_messages if msg["role"] in ("user", "assistant") + msg + for msg in context_messages + if msg["role"] in ("user", "assistant") + and not _is_memory_injection(msg) ] unsent_messages = storable_messages[self._messages_sent_count :] diff --git a/packages/pipecat-sdk-python/tests/test_no_memory_restorage.py b/packages/pipecat-sdk-python/tests/test_no_memory_restorage.py new file mode 100644 index 000000000..b88279d1a --- /dev/null +++ b/packages/pipecat-sdk-python/tests/test_no_memory_restorage.py @@ -0,0 +1,225 @@ +"""Regression tests: injected blocks must not be re-stored. + +LLMContext.get_messages() returns the context's live message list (the +enhancement code relies on that to mutate it in place). The frame handler +used to snapshot that same reference *before* injecting the memory block, +so the injected user-message showed up in storable_messages and was written +back to Supermemory as a brand-new memory — recalled and re-stored again on +every following turn. +""" + +from __future__ import annotations + +import sys +import types +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock + + +def _install_test_stubs() -> None: + if "loguru" not in sys.modules: + loguru_module = types.ModuleType("loguru") + + class _Logger: + def warning(self, *_args, **_kwargs): + return None + + def error(self, *_args, **_kwargs): + return None + + loguru_module.logger = _Logger() + sys.modules["loguru"] = loguru_module + + if "pydantic" not in sys.modules: + pydantic_module = types.ModuleType("pydantic") + + class BaseModel: + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def Field(*, default=None, **_kwargs): + return default + + pydantic_module.BaseModel = BaseModel + pydantic_module.Field = Field + sys.modules["pydantic"] = pydantic_module + + if "pipecat" not in sys.modules: + pipecat_module = types.ModuleType("pipecat") + sys.modules["pipecat"] = pipecat_module + + frames_module = types.ModuleType("pipecat.frames.frames") + + class Frame: # pragma: no cover - import stub + pass + + class InputAudioRawFrame: # pragma: no cover - import stub + pass + + class LLMContextFrame: # pragma: no cover - import stub + pass + + class LLMMessagesFrame: # pragma: no cover - import stub + pass + + frames_module.Frame = Frame + frames_module.InputAudioRawFrame = InputAudioRawFrame + frames_module.LLMContextFrame = LLMContextFrame + frames_module.LLMMessagesFrame = LLMMessagesFrame + + llm_context_module = types.ModuleType( + "pipecat.processors.aggregators.llm_context" + ) + + class LLMContext: # pragma: no cover - import stub + pass + + llm_context_module.LLMContext = LLMContext + + openai_context_module = types.ModuleType( + "pipecat.processors.aggregators.openai_llm_context" + ) + + class OpenAILLMContextFrame: # pragma: no cover - import stub + pass + + openai_context_module.OpenAILLMContextFrame = OpenAILLMContextFrame + + frame_processor_module = types.ModuleType( + "pipecat.processors.frame_processor" + ) + + class FrameDirection: # pragma: no cover - import stub + pass + + class FrameProcessor: + def __init__(self, *args, **kwargs): + return None + + frame_processor_module.FrameDirection = FrameDirection + frame_processor_module.FrameProcessor = FrameProcessor + + sys.modules["pipecat.frames.frames"] = frames_module + sys.modules["pipecat.processors.aggregators.llm_context"] = llm_context_module + sys.modules[ + "pipecat.processors.aggregators.openai_llm_context" + ] = openai_context_module + sys.modules["pipecat.processors.frame_processor"] = frame_processor_module + + # process_frame calls super().process_frame(); give whichever stub is + # installed (this file's or test_empty_profile's) an async no-op. + fp = sys.modules["pipecat.processors.frame_processor"] + if not hasattr(fp.FrameProcessor, "process_frame"): + + async def _noop_process_frame(self, _frame, _direction): + return None + + fp.FrameProcessor.process_frame = _noop_process_frame + + +_install_test_stubs() + +from pipecat.frames.frames import LLMContextFrame + +from supermemory_pipecat.service import ( + MEMORY_TAG_START, + SupermemoryPipecatService, + _is_memory_injection, +) + + +class _FakeLLMContext: + """Mimics pipecat's LLMContext: get_messages() returns the live list.""" + + def __init__(self, messages): + self._messages = messages + + def get_messages(self): + return self._messages + + def add_message(self, message): + self._messages.append(message) + + +class _MockSupermemoryClient: + def __init__(self, response): + self.profile = AsyncMock(return_value=response) + + +def _profile_response(): + return SimpleNamespace( + profile=SimpleNamespace(static=["User likes Python"], dynamic=[]), + search_results=None, + ) + + +def _make_service(): + service = SupermemoryPipecatService(api_key="mock_key", user_id="user_123") + service._supermemory_client = _MockSupermemoryClient(_profile_response()) + service._store_messages = AsyncMock() + service.push_frame = AsyncMock() + return service + + +def _frame_for(context): + frame = LLMContextFrame() + frame.context = context + return frame + + +class TestNoMemoryRestorage(unittest.IsolatedAsyncioTestCase): + async def test_injected_memory_block_is_not_stored(self) -> None: + service = _make_service() + context = _FakeLLMContext([{"role": "user", "content": "Hello"}]) + + await service.process_frame(_frame_for(context), direction=None) + + # The memory block was injected into the live context... + injected = [m for m in context.get_messages() if _is_memory_injection(m)] + self.assertEqual(len(injected), 1) + + # ...but only the real user message was queued for storage. + service._store_messages.assert_called_once() + stored = service._store_messages.call_args.args[0] + self.assertEqual(stored, [{"role": "user", "content": "Hello"}]) + + async def test_previous_turn_injection_is_filtered_from_storage(self) -> None: + service = _make_service() + service._messages_sent_count = 1 # "Hello" already stored last turn + context = _FakeLLMContext( + [ + {"role": "user", "content": "Hello"}, + {"role": "user", "content": f"{MEMORY_TAG_START}\nrecalled\n"}, + {"role": "assistant", "content": "Hi there"}, + {"role": "user", "content": "What next?"}, + ] + ) + + await service.process_frame(_frame_for(context), direction=None) + + service._store_messages.assert_called_once() + stored = service._store_messages.call_args.args[0] + self.assertEqual( + stored, + [ + {"role": "assistant", "content": "Hi there"}, + {"role": "user", "content": "What next?"}, + ], + ) + for msg in stored: + self.assertNotIn(MEMORY_TAG_START, msg["content"]) + + async def test_is_memory_injection_tolerates_non_string_content(self) -> None: + self.assertFalse(_is_memory_injection({"role": "user", "content": None})) + self.assertFalse(_is_memory_injection({"role": "user", "content": ["parts"]})) + self.assertTrue( + _is_memory_injection( + {"role": "user", "content": f"{MEMORY_TAG_START} recalled"} + ) + ) + + +if __name__ == "__main__": + unittest.main()