diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
index adf2ab411..c52b51646 100644
--- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
+++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
@@ -415,8 +415,13 @@ def _enhance_context_with_memories(
if system_idx is not None:
existing_content = messages[system_idx].get("content", "")
if MEMORY_TAG_PATTERN.search(existing_content):
+ # A callable replacement keeps the memory text literal. As a
+ # plain string, re.sub reads backslash escapes in it as group
+ # references and raises on real memories such as a Windows
+ # path ("bad escape \\U") or a regex fact ("invalid group
+ # reference 1").
messages[system_idx]["content"] = MEMORY_TAG_PATTERN.sub(
- tagged_memory, existing_content
+ lambda _match: tagged_memory, existing_content
)
else:
messages[system_idx]["content"] = f"{existing_content}\n\n{tagged_memory}"
diff --git a/packages/pipecat-sdk-python/tests/test_empty_profile.py b/packages/pipecat-sdk-python/tests/test_empty_profile.py
index 184cff6ce..aae328c32 100644
--- a/packages/pipecat-sdk-python/tests/test_empty_profile.py
+++ b/packages/pipecat-sdk-python/tests/test_empty_profile.py
@@ -154,3 +154,52 @@ def add_message(self, message):
self.assertTrue(
any(fact in message.get("content", "") for message in context.messages)
)
+
+ def test_system_injection_replaces_stale_block_with_backslash_memory(self) -> None:
+ """A memory containing backslashes must not be read as a regex template.
+
+ re.sub() expands escapes in a *string* replacement, so a Windows path or
+ a group-reference-looking fact used to raise re.error and abort the turn.
+ """
+ backslash = chr(92)
+ fact = "User's repo is at C:" + backslash + "Users" + backslash + "alice"
+ service = SupermemoryPipecatService(
+ api_key="mock_key",
+ user_id="user-123",
+ session_id="conversation-456",
+ params=SupermemoryPipecatService.InputParams(inject_mode="system"),
+ )
+ newline = chr(10)
+ stale = newline.join(["", "Stale fact", ""])
+
+ class Context:
+ def __init__(self):
+ self.messages = [
+ {
+ "role": "system",
+ "content": "You are helpful." + newline + newline + stale,
+ },
+ {"role": "user", "content": "Where is my repo?"},
+ ]
+
+ def get_messages(self):
+ return self.messages
+
+ def add_message(self, message):
+ self.messages.append(message)
+
+ context = Context()
+ service._enhance_context_with_memories(
+ context,
+ "Where is my repo?",
+ {
+ "profile": {"static": [fact], "dynamic": []},
+ "search_results": [],
+ },
+ )
+
+ system_content = context.messages[0]["content"]
+ self.assertIn(fact, system_content)
+ self.assertNotIn("Stale fact", system_content)
+ self.assertIn("You are helpful.", system_content)
+ self.assertEqual(system_content.count(""), 1)