From 8e11c1f786359009085de47fea2d1c4ea520c966 Mon Sep 17 00:00:00 2001
From: Agnik47 <140933190+Agnik47@users.noreply.github.com>
Date: Fri, 4 Sep 2026 04:21:21 +0530
Subject: [PATCH] fix(pipecat): keep memory text literal when replacing the
 injected block

_enhance_context_with_memories refreshes the system message by calling
MEMORY_TAG_PATTERN.sub(tagged_memory, existing_content). Passing the
memory text as a string replacement makes re.sub parse it as a template,
so backslash sequences in ordinary memories are expanded rather than
inserted.

Two failures follow, both on the second and later turns, once a block
exists to replace:

  "User's repo is at C:\Users\alice"  -> re.error: bad escape \U
  "User writes \1 for a capture group" -> re.error: invalid group reference 1

The exception propagates out of context enhancement and aborts the turn.
A callable replacement inserts the text verbatim and cannot be parsed as
a template.

Adds a regression test covering the Windows-path case, which fails on
main with the production error.
---
 .../src/supermemory_pipecat/service.py        |  7 ++-
 .../tests/test_empty_profile.py               | 49 +++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

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(["<user_memories>", "Stale fact", "</user_memories>"])
+
+        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("<user_memories>"), 1)