diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py index 16230cdf5..a9a0785c0 100644 --- a/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py +++ b/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 diff --git a/packages/cartesia-sdk-python/tests/test_single_agent_run.py b/packages/cartesia-sdk-python/tests/test_single_agent_run.py new file mode 100644 index 000000000..d2786db4c --- /dev/null +++ b/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()