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()