diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index b8d2db4ef3c9..a525622e99f4 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -202,6 +202,7 @@ def __init__( self._async_queue: queue.Queue | None = None self._async_thread: threading.Thread | None = None self._async_thread_lock = threading.Lock() + self._flush_lock = threading.Lock() if write_frequency == "async": self._async_queue = queue.Queue() @@ -630,7 +631,12 @@ def get_or_create(self, key: str) -> HonchoSession: return session def _flush_session(self, session: HonchoSession) -> bool: - """Internal: write unsynced messages to Honcho synchronously.""" + """Internal: serialize writes of unsynced messages to Honcho.""" + with self._flush_lock: + return self._flush_session_locked(session) + + def _flush_session_locked(self, session: HonchoSession) -> bool: + """Write unsynced messages while the flush lock is held.""" if not session.messages: return True diff --git a/tests/honcho_plugin/test_async_memory.py b/tests/honcho_plugin/test_async_memory.py index a6dddf4c292a..c86c1becd8f6 100644 --- a/tests/honcho_plugin/test_async_memory.py +++ b/tests/honcho_plugin/test_async_memory.py @@ -316,6 +316,47 @@ def test_shutdown_without_started_thread_is_noop(self, make_manager): mgr.shutdown() assert mgr._async_thread is None + def test_concurrent_flushes_do_not_duplicate_messages(self, make_manager): + mgr = make_manager(write_frequency="turn") + session = _make_session(key="concurrent-flush") + session.add_message("user", "only once") + mgr._peers_cache[session.user_peer_id] = MagicMock() + mgr._peers_cache[session.assistant_peer_id] = MagicMock() + honcho_session = MagicMock() + mgr._sessions_cache[session.honcho_session_id] = honcho_session + + first_upload_started = threading.Event() + second_upload_started = threading.Event() + release_upload = threading.Event() + upload_count = 0 + count_lock = threading.Lock() + + def blocking_add_messages(_messages): + nonlocal upload_count + with count_lock: + upload_count += 1 + if upload_count == 1: + first_upload_started.set() + else: + second_upload_started.set() + release_upload.wait(timeout=1) + + honcho_session.add_messages.side_effect = blocking_add_messages + first = threading.Thread(target=lambda: mgr._flush_session(session), daemon=True) + second = threading.Thread(target=lambda: mgr._flush_session(session), daemon=True) + first.start() + assert first_upload_started.wait(timeout=1) + second.start() + + assert not second_upload_started.wait(timeout=0.05) + release_upload.set() + first.join(timeout=1) + second.join(timeout=1) + + assert not first.is_alive() + assert not second.is_alive() + assert honcho_session.add_messages.call_count == 1 + # --------------------------------------------------------------------------- # async retry on failure