fix(cartesia): run the wrapped agent exactly once per event

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

@@ -477,8 +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
+            # Guard only the enrichment and storage work so a memory failure
+            # cannot stop the agent, and keep the agent call out of the guard
+            # so its own errors are not retried into a duplicate run.
+            try:
                 logger.info("[Supermemory] Processing UserTurnEnded event")
                 event, memory_context = await self._enrich_event_with_memories(event)
 
@@ -505,15 +509,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] Error in memory enrichment: {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/tests/test_single_agent_run.py

@@ -0,0 +1,131 @@
+"""Regression tests: the wrapped agent must run exactly once per event.
+
+A mid-stream agent error must propagate instead of re-running the agent, and
+a memory-enrichment failure must still let the agent run exactly once.
+"""
+
+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)
+
+        # Each chunk delivered once and the agent ran once: no duplicate run.
+        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())
+        ]
+
+        # Memory failure is absorbed; the agent still runs, exactly once.
+        self.assertEqual(outputs, ["chunk-1", "chunk-2"])
+        self.assertEqual(inner.run_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()