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

process() wrapped both the memory enrichment and the agent streaming
loop in a single try/except whose fallback re-invoked
self.agent.process(). The fallback was meant to keep the agent running
when memory enrichment failed, but because the agent's own iteration
lived inside the same try, an agent error raised mid-stream, after
chunks had already been yielded to the caller, re-ran the entire agent
from scratch: the caller received a second full response concatenated
onto the partial one, plus a duplicate billable LLM call.

Scope the guard to the memory-enrichment/injection/storage block only,
and run the agent iteration exactly once after it. Enrichment failures
are logged and absorbed as before; agent errors now propagate to the
caller instead of triggering a duplicate run.

Regression tests stream from a fake inner agent that dies mid-stream:
each chunk is delivered exactly once, the agent's run counter stays at
1, and the error propagates (both tests fail against the previous
implementation with run_count == 2). A third test pins the intended
behaviour that enrichment failures still let the agent run once.

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

@@ -477,8 +477,12 @@ class SupermemoryCartesiaAgent:
         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 @@ class SupermemoryCartesiaAgent:
                         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()