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.
This commit is contained in:
abhay-codes07 2026-07-13 10:25:36 +05:30
parent 2cebe81512
commit 8c84e0fe45
No known key found for this signature in database
2 changed files with 149 additions and 12 deletions

View file

@ -387,8 +387,14 @@ class SupermemoryCartesiaAgent:
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 @@ class SupermemoryCartesiaAgent:
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."""

View file

@ -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()