From c3295ab3b9cda0b725f3dcaab64939734bd6da0a Mon Sep 17 00:00:00 2001 From: abhay-codes07 Date: Mon, 13 Jul 2026 10:25:36 +0530 Subject: [PATCH] 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. --- .../src/supermemory_cartesia/agent.py | 21 +-- .../tests/test_single_agent_run.py | 131 ++++++++++++++++++ 2 files changed, 142 insertions(+), 10 deletions(-) create mode 100644 packages/cartesia-sdk-python/tests/test_single_agent_run.py 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()