From 8c84e0fe45e2026879e12ecb31cf7f5694755459 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 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 26 ++-- .../tests/test_single_agent_run.py | 135 ++++++++++++++++++ 2 files changed, 149 insertions(+), 12 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 9019193e8..3711432c9 100644 --- a/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py +++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py @@ -387,8 +387,14 @@ async def process(self, env: Any, event: Event) -> AsyncGenerator[Event, None]: Yields: Output events from the wrapped agent. """ - try: - if type(event).__name__ == "UserTurnEnded": + # Memory enrichment must never prevent the agent from running, so it + # gets its own guard. The agent iteration stays outside of it: with + # the old whole-body try/except, an agent error raised mid-stream + # (after some chunks were already yielded) triggered a fallback that + # re-ran the entire agent — duplicating the visible response and the + # LLM call. + if type(event).__name__ == "UserTurnEnded": + try: logger.info("[Supermemory] Processing UserTurnEnded event") event, memory_context = await self._enrich_event_with_memories(event) @@ -433,17 +439,13 @@ async def process(self, env: Any, event: Event) -> AsyncGenerator[Event, None]: self._background_tasks.add(task) task.add_done_callback(self._background_tasks.discard) self._messages_sent_count = 1 # CRITICAL: Increment counter to prevent duplicate storage + except Exception as e: + logger.error(f"[Supermemory] Error in memory enrichment: {e}") - async for output in self.agent.process(env, event): - 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.agent.process(env, event): - yield output + # Run the wrapped agent exactly once; its errors propagate to the + # caller instead of triggering a duplicate run. + async for output in self.agent.process(env, event): + yield output def reset_memory_tracking(self) -> None: """Reset memory tracking for a new conversation.""" 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..9cc06d7a7 --- /dev/null +++ b/packages/cartesia-sdk-python/tests/test_single_agent_run.py @@ -0,0 +1,135 @@ +"""Regression tests: the wrapped agent must run exactly once per event. + +The whole body of process() used to sit in one try/except whose fallback +re-invoked self.agent.process(). An agent error raised mid-stream — after +chunks had already been yielded to the caller — therefore re-ran the entire +agent: duplicated visible output and a duplicate LLM call. Memory-enrichment +failures are the only thing the fallback was meant to absorb. +""" + +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) + + # The caller saw each chunk exactly once and the agent ran once — + # no duplicated response, no duplicate LLM call. + 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()