fix: crash when profile search returns hits (Cartesia + Pipecat)

packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py

@@ -49,17 +49,34 @@ def format_relative_time(iso_timestamp: str) -> str:
         return ""
 
 
+def _result_to_dict(result: Any) -> Dict[str, Any]:
+    """Normalize a single search result into a plain dict.
+
+    The Supermemory SDK returns search results as Pydantic model objects, not
+    dicts, so calling ``.get(...)`` on them raises AttributeError. Convert
+    models via ``model_dump`` (keeping API field names like ``updatedAt``) and
+    pass existing dicts through unchanged.
+    """
+    if isinstance(result, dict):
+        return result
+    model_dump = getattr(result, "model_dump", None)
+    if callable(model_dump):
+        return model_dump(by_alias=True)
+    return {}
+
+
 def deduplicate_memories(
     static: List[str],
     dynamic: List[str],
-    search_results: List[Dict[str, Any]],
+    search_results: List[Any],
 ) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]:
     """Deduplicate memories. Priority: static > dynamic > search.
 
     Args:
         static: List of static memory strings.
         dynamic: List of dynamic memory strings.
-        search_results: List of search result dicts with 'memory' and 'updatedAt'.
+        search_results: List of search results (SDK model objects or dicts)
+            carrying 'memory' and 'updatedAt'.
     """
     seen = set()
 
@@ -71,9 +88,10 @@ def unique_strings(memories: List[str]) -> List[str]:
                 out.append(m)
         return out
 
-    def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    def unique_search(results: List[Any]) -> List[Dict[str, Any]]:
         out = []
         for r in results:
+            r = _result_to_dict(r)
             memory = r.get("memory", "")
             if memory and memory not in seen:
                 seen.add(memory)

packages/cartesia-sdk-python/tests/test_search_result_models.py

@@ -0,0 +1,55 @@
+"""Regression tests: profile search results arrive as SDK model objects.
+
+The Supermemory SDK returns ``response.search_results.results`` as Pydantic
+model objects, not dicts. Calling ``.get("memory")`` on a model raises
+AttributeError, which crashed ``deduplicate_memories`` on every non-empty
+search (the default ``mode="full"`` path). These tests reproduce that path
+with a dict-less model stand-in.
+"""
+
+import unittest
+
+from supermemory_cartesia.utils import deduplicate_memories, format_memories_to_text
+
+
+class _FakeSearchResult:
+    """Mimics a Supermemory SDK search result: attribute access and
+    ``model_dump()`` but deliberately no dict ``.get()``."""
+
+    def __init__(self, memory, updated_at=None):
+        self.memory = memory
+        self.updatedAt = updated_at
+
+    def model_dump(self, by_alias=False):
+        data = {"memory": self.memory}
+        if self.updatedAt is not None:
+            data["updatedAt"] = self.updatedAt
+        return data
+
+
+class TestSearchResultModels(unittest.TestCase):
+    def test_model_results_do_not_crash_and_dedupe(self):
+        results = [
+            _FakeSearchResult("likes python", "2020-01-01T00:00:00Z"),
+            _FakeSearchResult("likes python"),  # duplicate, dropped
+            _FakeSearchResult("prefers async"),
+        ]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertEqual(
+            [r["memory"] for r in dedup["search_results"]],
+            ["likes python", "prefers async"],
+        )
+
+    def test_model_results_render_to_text(self):
+        results = [_FakeSearchResult("likes python", "2020-01-01T00:00:00Z")]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertIn("likes python", format_memories_to_text(dedup))
+
+    def test_dict_results_still_supported(self):
+        results = [{"memory": "from dict"}]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertEqual([r["memory"] for r in dedup["search_results"]], ["from dict"])
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py

@@ -49,17 +49,34 @@ def format_relative_time(iso_timestamp: str) -> str:
         return ""
 
 
+def _result_to_dict(result: Any) -> Dict[str, Any]:
+    """Normalize a single search result into a plain dict.
+
+    The Supermemory SDK returns search results as Pydantic model objects, not
+    dicts, so calling ``.get(...)`` on them raises AttributeError. Convert
+    models via ``model_dump`` (keeping API field names like ``updatedAt``) and
+    pass existing dicts through unchanged.
+    """
+    if isinstance(result, dict):
+        return result
+    model_dump = getattr(result, "model_dump", None)
+    if callable(model_dump):
+        return model_dump(by_alias=True)
+    return {}
+
+
 def deduplicate_memories(
     static: List[str],
     dynamic: List[str],
-    search_results: List[Dict[str, Any]],
+    search_results: List[Any],
 ) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]:
     """Deduplicate memories. Priority: static > dynamic > search.
 
     Args:
         static: List of static memory strings.
         dynamic: List of dynamic memory strings.
-        search_results: List of search result dicts with 'memory' and 'updatedAt'.
+        search_results: List of search results (SDK model objects or dicts)
+            carrying 'memory' and 'updatedAt'.
     """
     seen = set()
 
@@ -71,9 +88,10 @@ def unique_strings(memories: List[str]) -> List[str]:
                 out.append(m)
         return out
 
-    def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    def unique_search(results: List[Any]) -> List[Dict[str, Any]]:
         out = []
         for r in results:
+            r = _result_to_dict(r)
             memory = r.get("memory", "")
             if memory and memory not in seen:
                 seen.add(memory)

packages/pipecat-sdk-python/tests/test_search_result_models.py

@@ -0,0 +1,55 @@
+"""Regression tests: profile search results arrive as SDK model objects.
+
+The Supermemory SDK returns ``response.search_results.results`` as Pydantic
+model objects, not dicts. Calling ``.get("memory")`` on a model raises
+AttributeError, which crashed ``deduplicate_memories`` on every non-empty
+search (the default ``mode="full"`` path). These tests reproduce that path
+with a dict-less model stand-in.
+"""
+
+import unittest
+
+from supermemory_pipecat.utils import deduplicate_memories, format_memories_to_text
+
+
+class _FakeSearchResult:
+    """Mimics a Supermemory SDK search result: attribute access and
+    ``model_dump()`` but deliberately no dict ``.get()``."""
+
+    def __init__(self, memory, updated_at=None):
+        self.memory = memory
+        self.updatedAt = updated_at
+
+    def model_dump(self, by_alias=False):
+        data = {"memory": self.memory}
+        if self.updatedAt is not None:
+            data["updatedAt"] = self.updatedAt
+        return data
+
+
+class TestSearchResultModels(unittest.TestCase):
+    def test_model_results_do_not_crash_and_dedupe(self):
+        results = [
+            _FakeSearchResult("likes python", "2020-01-01T00:00:00Z"),
+            _FakeSearchResult("likes python"),  # duplicate, dropped
+            _FakeSearchResult("prefers async"),
+        ]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertEqual(
+            [r["memory"] for r in dedup["search_results"]],
+            ["likes python", "prefers async"],
+        )
+
+    def test_model_results_render_to_text(self):
+        results = [_FakeSearchResult("likes python", "2020-01-01T00:00:00Z")]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertIn("likes python", format_memories_to_text(dedup))
+
+    def test_dict_results_still_supported(self):
+        results = [{"memory": "from dict"}]
+        dedup = deduplicate_memories(static=[], dynamic=[], search_results=results)
+        self.assertEqual([r["memory"] for r in dedup["search_results"]], ["from dict"])
+
+
+if __name__ == "__main__":
+    unittest.main()