fix(honcho): serialize concurrent message flushes

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
 

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