fix(python): community fixes for the openai, cartesia and pipecat wrappers

apps/docs/integrations/pipecat.mdx

@@ -28,8 +28,7 @@ You can obtain an API key from [console.supermemory.ai](https://console.supermem
 Supermemory integration is provided through the `SupermemoryPipecatService` class in Pipecat:
 
 ```python
-from supermemory_pipecat import SupermemoryPipecatService
-from supermemory_pipecat.service import InputParams
+from supermemory_pipecat import InputParams, SupermemoryPipecatService
 
 memory = SupermemoryPipecatService(
     api_key=os.getenv("SUPERMEMORY_API_KEY"),
@@ -157,8 +156,7 @@ from pipecat.transports.websocket.fastapi import (
     FastAPIWebsocketTransport,
 )
 
-from supermemory_pipecat import SupermemoryPipecatService
-from supermemory_pipecat.service import InputParams
+from supermemory_pipecat import InputParams, SupermemoryPipecatService
 
 app = FastAPI()
 

packages/cartesia-sdk-python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "supermemory-cartesia"
-version = "0.1.4"
+version = "0.1.5"
 description = "Supermemory integration for Cartesia Line - memory-enhanced voice agents"
 readme = "README.md"
 license = "MIT"
@@ -13,7 +13,7 @@ authors = [
     { name = "Supermemory", email = "support@supermemory.ai" }
 ]
 keywords = [
-    "supermemory>=3.16.0,<5",
+    "supermemory",
     "cartesia",
     "line",
     "memory",
@@ -32,7 +32,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Artificial Intelligence",
 ]
 dependencies = [
-    "supermemory>=3.16.0",
+    "supermemory>=3.16.0,<5",
     "cartesia-line>=0.2.0,<0.3.0",
     "pydantic>=2.10.0",
     "loguru>=0.7.3",

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

@@ -56,7 +56,7 @@
     __version__ = version("supermemory-cartesia")
 except PackageNotFoundError:
     # Source checkouts do not have installed distribution metadata.
-    __version__ = "0.1.3"
+    __version__ = "0.1.5"
 
 __all__ = [
     # Main agent

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

@@ -477,12 +477,12 @@ async def process(self, env: Any, event: Event) -> AsyncGenerator[Event, None]:
         Yields:
             Output events from the wrapped agent.
         """
-        try:
-            if type(event).__name__ == "UserTurnEnded":
+        if type(event).__name__ == "UserTurnEnded":
+            memory_context = None
+            try:
                 logger.info("[Supermemory] Processing UserTurnEnded event")
                 event, memory_context = await self._enrich_event_with_memories(event)
 
-                # Store conversation in background
                 if hasattr(event, 'history') and event.history:
                     new_messages = self._new_history_messages(event.history)
                     if new_messages:
@@ -505,15 +505,12 @@ async def process(self, env: Any, event: Event) -> AsyncGenerator[Event, None]:
                         task = asyncio.create_task(self._store_messages(new_messages))
                         self._background_tasks.add(task)
                         task.add_done_callback(self._background_tasks.discard)
+            except Exception as e:
+                logger.error(f"[Supermemory] Failed to enrich or queue the conversation turn: {e}")
 
-                async for output in self._process_agent(env, event, memory_context):
-                    yield output
-            else:
-                async for output in self.agent.process(env, event):
-                    yield output
-
-        except Exception as e:
-            logger.error(f"[Supermemory] Error in process: {e}")
+            async for output in self._process_agent(env, event, memory_context):
+                yield output
+        else:
             async for output in self.agent.process(env, event):
                 yield output
 

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

@@ -5,11 +5,12 @@
 from typing import Any, Dict, List, Union
 
 
-def get_last_user_message(messages: List[Dict[str, str]]) -> str | None:
+def get_last_user_message(messages: List[Dict[str, Any]]) -> str | None:
     """Extract the last user message content from a list of messages."""
     for msg in reversed(messages):
-        if msg["role"] == "user":
-            return msg["content"]
+        content = msg.get("content")
+        if msg.get("role") == "user" and isinstance(content, str):
+            return content
     return None
 
 
@@ -70,6 +71,17 @@ def _field(item: Any, *names: str, default: Any = None) -> Any:
     return default
 
 
+def _memory_text(item: Any) -> str:
+    """First non-empty memory field; `_field` alone would stop at an empty `memory`."""
+    if isinstance(item, str):
+        return item.strip()
+    for name in ("memory", "chunk", "content"):
+        value = _field(item, name)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return ""
+
+
 _MEMORY_DATE_PREFIX = re.compile(
     r"^\s*(?:\[recent\]\s*)?(?:\[\d{4}-\d{2}-\d{2}\]\s*)?",
     re.IGNORECASE,
@@ -124,10 +136,7 @@ def unique_search(results: List[Any]) -> List[Any]:
         out = []
         for r in results:
             # v4 search.memories/hybrid uses `memory` or `chunk`.
-            memory = _field(r, "memory", "chunk", "content", default="")
-            if not isinstance(memory, str):
-                memory = ""
-            memory = memory.strip()
+            memory = _memory_text(r)
             key = _memory_key(memory)
             if key and key not in seen:
                 seen.add(key)
@@ -174,7 +183,7 @@ def format_memories_to_text(
                 lines.append(f"- {item}")
                 continue
 
-            memory = _field(item, "memory", "chunk", "content", default="")
+            memory = _memory_text(item)
             updated_at = _field(item, "updatedAt", "updated_at", default="")
             time_str = format_relative_time(updated_at) if updated_at else ""
             if time_str:

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

@@ -92,6 +92,38 @@ def test_query_mode_keeps_search_fact_also_present_in_profile(self) -> None:
         self.assertIsNotNone(context)
         self.assertIn(fact, context)
 
+    async def test_process_does_not_retry_agent_after_output(self) -> None:
+        class UserTurnEnded:
+            pass
+
+        class FailingAfterOutputAgent:
+            def __init__(self) -> None:
+                self.calls = 0
+
+            async def process(self, _env, _event, **_kwargs):
+                self.calls += 1
+                yield "partial"
+                raise RuntimeError("agent failed after output")
+
+        wrapped_agent = FailingAfterOutputAgent()
+        agent = SupermemoryCartesiaAgent(
+            agent=wrapped_agent,
+            api_key="mock_key",
+            container_tag="user-123",
+            custom_id="conversation-456",
+            add_memory="never",
+        )
+        event = UserTurnEnded()
+        agent._enrich_event_with_memories = AsyncMock(return_value=(event, None))
+
+        outputs = []
+        with self.assertRaisesRegex(RuntimeError, "agent failed after output"):
+            async for output in agent.process(None, event):
+                outputs.append(output)
+
+        self.assertEqual(outputs, ["partial"])
+        self.assertEqual(wrapped_agent.calls, 1)
+
 
 if __name__ == "__main__":
     unittest.main()

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

@@ -0,0 +1,125 @@
+"""Regression tests: the wrapped agent must run exactly once per event."""
+
+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 info(self, *_args, **_kwargs):
+                return None
+
+            def debug(self, *_args, **_kwargs):
+                return None
+
+            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
+
+
+_install_test_stubs()
+
+from supermemory_cartesia.agent import SupermemoryCartesiaAgent
+
+
+class _StreamingAgent:
+    """Inner agent that yields chunks; optionally dies mid-stream."""
+
+    def __init__(self, fail_after_chunks: bool = False):
+        self.fail_after_chunks = fail_after_chunks
+        self.run_count = 0
+
+    async def process(self, env, event):
+        self.run_count += 1
+        yield "chunk-1"
+        yield "chunk-2"
+        if self.fail_after_chunks:
+            raise RuntimeError("stream dropped")
+
+
+def _wrap(inner):
+    return SupermemoryCartesiaAgent(
+        agent=inner,
+        api_key="mock_key",
+        container_tag="user-123",
+        custom_id="conversation-456",
+    )
+
+
+def _user_turn_ended_event():
+    return type("UserTurnEnded", (), {})()
+
+
+class TestSingleAgentRun(unittest.IsolatedAsyncioTestCase):
+    async def test_mid_stream_agent_error_is_not_retried(self) -> None:
+        inner = _StreamingAgent(fail_after_chunks=True)
+        wrapper = _wrap(inner)
+
+        outputs = []
+        with self.assertRaises(RuntimeError):
+            async for out in wrapper.process(None, SimpleNamespace()):
+                outputs.append(out)
+
+        self.assertEqual(outputs, ["chunk-1", "chunk-2"])
+        self.assertEqual(inner.run_count, 1)
+
+    async def test_mid_stream_error_on_user_turn_is_not_retried(self) -> None:
+        inner = _StreamingAgent(fail_after_chunks=True)
+        wrapper = _wrap(inner)
+        wrapper._enrich_event_with_memories = AsyncMock(
+            side_effect=lambda event: (event, None)
+        )
+
+        outputs = []
+        with self.assertRaises(RuntimeError):
+            async for out in wrapper.process(None, _user_turn_ended_event()):
+                outputs.append(out)
+
+        self.assertEqual(outputs, ["chunk-1", "chunk-2"])
+        self.assertEqual(inner.run_count, 1)
+
+    async def test_enrichment_failure_still_runs_agent_once(self) -> None:
+        inner = _StreamingAgent()
+        wrapper = _wrap(inner)
+        wrapper._enrich_event_with_memories = AsyncMock(
+            side_effect=RuntimeError("enrichment exploded")
+        )
+
+        outputs = [
+            out async for out in wrapper.process(None, _user_turn_ended_event())
+        ]
+
+        self.assertEqual(outputs, ["chunk-1", "chunk-2"])
+        self.assertEqual(inner.run_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()