From ff29c27003ad68bd49169850efeed7ae72943e78 Mon Sep 17 00:00:00 2001 From: Erosika Date: Fri, 4 Sep 2026 15:40:17 -0400 Subject: [PATCH] fix(honcho): hold a per-session lock across select, send and the _synced flip _flush_session read the unsynced messages, posted them, and only then set _synced. in a short-lived run the async writer draining save()'s queue and the exit-time flush_all() both saw the same batch unsynced, so every turn of a one-shot run landed in honcho twice (#92458). each HonchoSession now carries an RLock that _flush_session holds around the whole select, send, mark sequence. the second flusher enters after the first marked the batch and sends nothing; different sessions still flush in parallel. the lock lives on the session object instead of a manager dict keyed by session id (#86094, #92787): it needs no eviction, and two flushers of one message list can never hold different locks. tests adapted from #86094 and #92787. --- plugins/memory/honcho/session.py | 10 ++- tests/honcho_plugin/test_async_memory.py | 101 +++++++++++++++++++++++ 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index b7e568754b..e64916c18f 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -47,6 +47,8 @@ class HonchoSession: created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) metadata: dict[str, Any] = field(default_factory=dict) + # Held by _flush_session across select, send and the _synced flip; lives with the message list it guards. + _flush_lock: threading.RLock = field(default_factory=threading.RLock, repr=False, compare=False) def add_message(self, role: str, content: str, **kwargs: Any) -> None: """Add a message to the local cache.""" @@ -344,7 +346,13 @@ class HonchoSessionManager(SessionAuthMixin, SessionPeersMixin, SessionContextMi excess -= 1 def _flush_session(self, session: HonchoSession) -> bool: - """Write unsynced messages to Honcho synchronously.""" + """Write unsynced messages to Honcho synchronously. The session's lock serializes flushers of + one message list: the async writer and an exit-time flush_all() both pick up the same batch + otherwise, and the second one enters here only after the first marked it synced.""" + with session._flush_lock: + return self._flush_session_locked(session) + + def _flush_session_locked(self, session: HonchoSession) -> bool: new_messages = [m for m in session.messages if not m.get("_synced")] if not new_messages: return True diff --git a/tests/honcho_plugin/test_async_memory.py b/tests/honcho_plugin/test_async_memory.py index 4a9f41aad9..4096929f21 100644 --- a/tests/honcho_plugin/test_async_memory.py +++ b/tests/honcho_plugin/test_async_memory.py @@ -601,3 +601,104 @@ class TestPrefetchCacheAccessors: assert mgr.pop_context_result("cli:test") == payload assert mgr.pop_context_result("cli:test") == {} + + +# --------------------------------------------------------------------------- +# concurrent flushes of one session send each batch once (#92458) +# --------------------------------------------------------------------------- + +class TestConcurrentFlushSession: + def _wire_remote(self, mgr, session, add_messages): + mgr._peers_cache[session.user_peer_id] = MagicMock() + mgr._peers_cache[session.assistant_peer_id] = MagicMock() + remote = MagicMock() + remote.add_messages.side_effect = add_messages + mgr._sessions_cache[session.honcho_session_id] = remote + return remote + + def test_racing_flushes_send_the_batch_once(self, make_manager): + mgr = make_manager(write_frequency="turn") + session = _make_session(key="race") + session.add_message("user", "only once") + upload_started = threading.Event() + release_upload = threading.Event() + + def blocking_add_messages(_messages): + upload_started.set() + release_upload.wait(timeout=2) + + remote = self._wire_remote(mgr, session, blocking_add_messages) + results = [] + first = threading.Thread(target=lambda: results.append(mgr._flush_session(session)), daemon=True) + second = threading.Thread(target=lambda: results.append(mgr._flush_session(session)), daemon=True) + first.start() + assert upload_started.wait(timeout=1) + second.start() + # The second flusher must be parked on the lock, not inside add_messages. + second.join(timeout=0.2) + assert second.is_alive() + release_upload.set() + first.join(timeout=2) + second.join(timeout=2) + + assert results == [True, True] + assert remote.add_messages.call_count == 1 + assert all(m["_synced"] for m in session.messages) + + def test_async_writer_and_exit_flush_send_the_batch_once(self, make_manager): + mgr = make_manager(write_frequency="async") + session = _make_session(key="oneshot") + session.add_message("user", "hello") + session.add_message("assistant", "hi") + with mgr._cache_lock: + mgr._cache[session.key] = session + upload_started = threading.Event() + release_upload = threading.Event() + + def blocking_add_messages(_messages): + upload_started.set() + release_upload.wait(timeout=2) + + remote = self._wire_remote(mgr, session, blocking_add_messages) + mgr.save(session) + assert upload_started.wait(timeout=2), "async writer never started the upload" + exit_flush = threading.Thread(target=mgr.flush_all, daemon=True) + exit_flush.start() + exit_flush.join(timeout=0.2) + assert exit_flush.is_alive() + release_upload.set() + exit_flush.join(timeout=2) + mgr.shutdown() + + assert remote.add_messages.call_count == 1 + assert all(m["_synced"] for m in session.messages) + + def test_independent_sessions_flush_in_parallel(self, make_manager): + mgr = make_manager(write_frequency="turn") + sessions = [_make_session(key="a", honcho_session_id="a"), _make_session(key="b", honcho_session_id="b")] + barrier = threading.Barrier(2) + results = [] + for session in sessions: + session.add_message("user", session.key) + self._wire_remote(mgr, session, lambda _messages: barrier.wait(timeout=1)) + threads = [threading.Thread(target=lambda s=s: results.append(mgr._flush_session(s)), daemon=True) for s in sessions] + for t in threads: + t.start() + for t in threads: + t.join(timeout=2) + + assert not any(t.is_alive() for t in threads) + assert results == [True, True] + + def test_same_session_flush_is_reentrant(self, make_manager): + mgr = make_manager(write_frequency="turn") + session = _make_session() + calls = [] + + def nested(current): + calls.append(current) + return mgr._flush_session(current) if len(calls) == 1 else True + + mgr._flush_session_locked = nested + assert mgr._flush_session(session) is True + assert len(calls) == 2