"""Tests for the async-memory Honcho improvements. Covers: - write_frequency parsing (async / turn / session / int) - resolve_session_name with session_title - HonchoSessionManager.save() routing per write_frequency - async writer thread lifecycle and retry - flush_all() drains pending messages - shutdown() joins the thread """ import json import logging import threading import time from unittest.mock import MagicMock, patch import pytest from plugins.memory.honcho.client import HonchoClientConfig from plugins.memory.honcho.session import ( HonchoSession, HonchoSessionManager, ) from plugins.memory.honcho.session_peers import HonchoPeerUnresolvedError # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_session(**kwargs) -> HonchoSession: return HonchoSession( key=kwargs.get("key", "cli:test"), user_peer_id=kwargs.get("user_peer_id", "eri"), assistant_peer_id=kwargs.get("assistant_peer_id", "hermes"), honcho_session_id=kwargs.get("honcho_session_id", "cli-test"), messages=kwargs.get("messages", []), ) # B8: managers are built ONLY through the make_manager fixture below. The old # helper constructed the manager first and swapped in a MagicMock afterwards - # the honcho property refreshes the client via get_honcho_client() on every # access, so the late mock never protected flush paths and test messages were # written to a live local Honcho (production incident, session cli-test). @pytest.fixture def make_manager(monkeypatch): """Factory: fake client is injected BEFORE the constructor, shutdown is guaranteed for every created manager (even on assertion failure).""" from plugins.memory.honcho import session as session_module client = MagicMock() monkeypatch.setattr(session_module, "get_honcho_client", lambda *a, **k: client) created = [] def _make( write_frequency="turn", *, runtime_user_peer_name=None, **cfg_kwargs, ) -> HonchoSessionManager: cfg = HonchoClientConfig( write_frequency=write_frequency, api_key="test-key", enabled=True, **cfg_kwargs, ) mgr = HonchoSessionManager( honcho=client, config=cfg, runtime_user_peer_name=runtime_user_peer_name, ) created.append(mgr) return mgr _make.client = client yield _make for mgr in created: mgr.shutdown() # --------------------------------------------------------------------------- # write_frequency parsing from config file # --------------------------------------------------------------------------- class TestWriteFrequencyParsing: def test_integer_frequency(self, tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps({"apiKey": "k", "writeFrequency": 5})) cfg = HonchoClientConfig.from_global_config(config_path=cfg_file) assert cfg.write_frequency == 5 def test_host_block_overrides_root(self, tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps({ "apiKey": "k", "writeFrequency": "turn", "hosts": {"hermes": {"writeFrequency": "session"}}, })) cfg = HonchoClientConfig.from_global_config(config_path=cfg_file) assert cfg.write_frequency == "session" # --------------------------------------------------------------------------- # resolve_session_name with session_title # --------------------------------------------------------------------------- class TestResolveSessionNameTitle: def test_manual_override_beats_title(self): cfg = HonchoClientConfig(sessions={"/my/project": "manual-name"}) result = cfg.resolve_session_name("/my/project", session_title="the-title") assert result == "manual-name" @pytest.mark.parametrize( ("session_strategy", "title_source", "expected"), [ ("per-directory", "llm", "dir"), ("per-directory", "derived", "dir"), ("per-repo", "llm", "repo-name"), ("per-repo", "derived", "repo-name"), ("global", "llm", "my-workspace"), ("global", "derived", "my-workspace"), ], ) def test_automatic_title_does_not_override_strategy( self, session_strategy, title_source, expected, ): cfg = HonchoClientConfig( session_strategy=session_strategy, workspace_id="my-workspace", ) with patch.object(HonchoClientConfig, "_git_repo_name", return_value="repo-name"): result = cfg.resolve_session_name( "/some/dir", session_title="generated-title", session_title_source=title_source, ) assert result == expected def test_title_sanitized(self): cfg = HonchoClientConfig() result = cfg.resolve_session_name("/some/dir", session_title="my project/name!") # trailing dashes stripped by .strip('-') assert result == "my-project-name" def test_none_title_falls_back_to_dirname(self): cfg = HonchoClientConfig() result = cfg.resolve_session_name("/some/dir", session_title=None) assert result == "dir" def test_empty_title_falls_back_to_dirname(self): cfg = HonchoClientConfig() result = cfg.resolve_session_name("/some/dir", session_title="") assert result == "dir" def test_per_session_uses_session_id(self): cfg = HonchoClientConfig(session_strategy="per-session") result = cfg.resolve_session_name( "/some/dir", session_title="generated-title", session_title_source="llm", session_id="20260309_175514_9797dd", ) assert result == "20260309_175514_9797dd" def test_gateway_key_beats_per_session_id(self): # Gateways keep per-chat isolation even in per-session. cfg = HonchoClientConfig(session_strategy="per-session") result = cfg.resolve_session_name( "/some/dir", session_title="explicit-title", session_title_source="user", gateway_session_key="agent:main:telegram:dm:42", session_id="20260309_175514_9797dd", ) assert result == "agent-main-telegram-dm-42" # --------------------------------------------------------------------------- # save() routing per write_frequency # --------------------------------------------------------------------------- class TestSaveRouting: def _make_session_with_message(self, mgr=None): sess = _make_session() sess.add_message("user", "hello") sess.add_message("assistant", "hi") if mgr: mgr._cache[sess.key] = sess return sess def test_turn_flushes_immediately(self, make_manager): mgr = make_manager(write_frequency="turn") sess = self._make_session_with_message(mgr) with patch.object(mgr, "_flush_session") as mock_flush: mgr.save(sess) mock_flush.assert_called_once_with(sess) def test_session_mode_does_not_flush(self, make_manager): mgr = make_manager(write_frequency="session") sess = self._make_session_with_message(mgr) with patch.object(mgr, "_flush_session") as mock_flush: mgr.save(sess) mock_flush.assert_not_called() def test_async_mode_enqueues(self, make_manager): mgr = make_manager(write_frequency="async") sess = self._make_session_with_message(mgr) with patch.object(mgr, "_flush_session") as mock_flush: mgr.save(sess) # flush_session should NOT be called synchronously mock_flush.assert_not_called() assert not mgr._async_queue.empty() def test_int_frequency_flushes_on_nth_turn(self, make_manager): mgr = make_manager(write_frequency=3) sess = self._make_session_with_message(mgr) with patch.object(mgr, "_flush_session") as mock_flush: mgr.save(sess) # turn 1 mgr.save(sess) # turn 2 assert mock_flush.call_count == 0 mgr.save(sess) # turn 3 assert mock_flush.call_count == 1 # --------------------------------------------------------------------------- # flush_all() # --------------------------------------------------------------------------- class TestFlushAll: def test_flushes_all_cached_sessions(self, make_manager): mgr = make_manager(write_frequency="session") s1 = _make_session(key="s1", honcho_session_id="s1") s2 = _make_session(key="s2", honcho_session_id="s2") s1.add_message("user", "a") s2.add_message("user", "b") mgr._cache = {"s1": s1, "s2": s2} with patch.object(mgr, "_flush_session") as mock_flush: mgr.flush_all() assert mock_flush.call_count == 2 def test_flush_all_drains_async_queue(self, make_manager): mgr = make_manager(write_frequency="async") sess = _make_session() sess.add_message("user", "pending") with patch.object(mgr, "_flush_session") as mock_flush: # Put the item AFTER the mock is installed so the background # writer thread (if it dequeues before flush_all) still hits # the mock rather than the real _flush_session. mgr._async_queue.put(sess) mgr.flush_all() # Called at least once for the queued item assert mock_flush.call_count >= 1 def test_flush_all_tolerates_errors(self, make_manager): mgr = make_manager(write_frequency="session") sess = _make_session() mgr._cache = {"key": sess} with patch.object(mgr, "_flush_session", side_effect=RuntimeError("oops")): # Should not raise mgr.flush_all() # --------------------------------------------------------------------------- # async writer thread lifecycle # --------------------------------------------------------------------------- class TestAsyncWriterThread: def test_thread_starts_lazily_on_first_enqueue(self, make_manager): # B8: constructing a manager must not spawn background work mgr = make_manager(write_frequency="async") assert mgr._async_queue is not None assert mgr._async_thread is None mgr.save(_make_session()) assert mgr._async_thread is not None assert mgr._async_thread.is_alive() mgr.shutdown() def test_no_thread_for_turn_mode(self, make_manager): mgr = make_manager(write_frequency="turn") assert mgr._async_thread is None assert mgr._async_queue is None def test_shutdown_joins_thread(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() assert mgr._async_thread.is_alive() mgr.shutdown() assert not mgr._async_thread.is_alive() def test_async_writer_calls_flush(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "async msg") flushed = [] flushed_event = threading.Event() def capture(session): flushed.append(session) flushed_event.set() return True mgr._flush_session = capture mgr._async_queue.put(sess) assert flushed_event.wait(timeout=10), "async writer never flushed" mgr.shutdown() assert len(flushed) == 1 assert flushed[0] is sess def test_shutdown_without_started_thread_is_noop(self, make_manager): mgr = make_manager(write_frequency="async") mgr.shutdown() assert mgr._async_thread is None def test_stop_async_writer_joins_thread_without_flushing(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "must not be written") with mgr._cache_lock: mgr._cache[sess.key] = sess flushed = [] mgr._flush_session = lambda session: flushed.append(session) or True thread = mgr._async_thread mgr.stop_async_writer() thread.join(timeout=10) assert not thread.is_alive() assert flushed == [] def test_stop_async_writer_without_started_thread_is_noop(self, make_manager): mgr = make_manager(write_frequency="async") mgr.stop_async_writer() assert mgr._async_thread is None # --------------------------------------------------------------------------- # async retry on failure # --------------------------------------------------------------------------- class TestStopAsyncWriterDrain: def test_items_queued_before_the_join_are_flushed(self, make_manager): mgr = make_manager("async") flushed = [] mgr._flush_session = lambda s: flushed.append(s.key) or True mgr._async_queue.put(_make_session(key="late")) mgr.stop_async_writer() assert flushed == ["late"] assert mgr._async_queue.empty() def test_save_after_the_writer_stopped_flushes_inline(self, make_manager): mgr = make_manager("async") flushed = [] mgr._flush_session = lambda s: flushed.append(s.key) or True mgr.stop_async_writer() mgr.save(_make_session(key="after")) assert flushed == ["after"] assert mgr._async_queue.empty() def _pending_session(self, mgr, uploads): session = _make_session(key="pending") session.add_message("user", "pending") mgr._cache["pending"] = session mgr._async_queue.put(session) mgr._flush_session = lambda s: uploads.append(s.key) or True mgr._flush_session_locked = lambda s: uploads.append(s.key) or True return session def test_shutdown_with_the_budget_spent_starts_no_upload_and_warns_once(self, make_manager, caplog): """The SDK has no per-call timeout, so the budget can only stop uploads from starting. With no time left, shutdown must not open one and must say what stayed behind.""" mgr = make_manager("async") uploads = [] session = self._pending_session(mgr, uploads) started = time.monotonic() with caplog.at_level(logging.WARNING, logger="plugins.memory.honcho"): mgr.shutdown(timeout=0) assert uploads == [] assert time.monotonic() - started < 2.0 assert mgr._async_queue.empty() assert session.messages[0].get("_synced") is None assert caplog.text.count("still unsynced") == 1 assert "1 message(s) in 1 session(s) still unsynced" in caplog.text def test_stop_async_writer_drains_only_within_its_timeout(self, make_manager, caplog): mgr = make_manager("async") uploads = [] self._pending_session(mgr, uploads) with caplog.at_level(logging.WARNING, logger="plugins.memory.honcho"): mgr.stop_async_writer(timeout=0) assert uploads == [] assert mgr._async_queue.empty() assert "1 message(s) in 1 session(s) still unsynced" in caplog.text class TestAsyncWriterRetry: def test_retries_once_on_failure(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "msg") call_count = [0] retry_done = threading.Event() def flaky_flush(session): call_count[0] += 1 if call_count[0] == 1: raise ConnectionError("network blip") retry_done.set() return True mgr._flush_session = flaky_flush with patch("time.sleep"): # skip the 2s sleep in retry mgr._async_queue.put(sess) assert retry_done.wait(timeout=10), "async writer never retried" mgr.shutdown() assert call_count[0] == 2 def test_does_not_retry_once_shutdown_began(self, make_manager): """The shutdown flush already attempts the session within its budget; a 2s sleep and a second upload from the writer would run past it.""" mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "msg") calls = [] failed = threading.Event() def failing_flush(session): calls.append(session) failed.set() return False mgr._flush_session = failing_flush mgr._shutting_down = True mgr._async_queue.put(sess) assert failed.wait(timeout=5), "async writer never picked up the batch" started = time.monotonic() mgr.stop_async_writer(timeout=5) assert time.monotonic() - started < 2.0 assert len(calls) == 1 def test_drops_after_two_failures(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "msg") call_count = [0] retry_done = threading.Event() def always_fail(session): call_count[0] += 1 if call_count[0] >= 2: retry_done.set() raise RuntimeError("always broken") mgr._flush_session = always_fail with patch("time.sleep"): mgr._async_queue.put(sess) assert retry_done.wait(timeout=10), "async writer never retried" mgr.shutdown() # Should have tried exactly twice (initial + one retry) and not crashed assert call_count[0] == 2 assert not mgr._async_thread.is_alive() def test_retries_when_flush_reports_failure(self, make_manager): mgr = make_manager(write_frequency="async") mgr._ensure_async_writer_locked() sess = _make_session() sess.add_message("user", "msg") call_count = [0] retry_done = threading.Event() def fail_then_succeed(session): call_count[0] += 1 if call_count[0] >= 2: retry_done.set() return call_count[0] > 1 mgr._flush_session = fail_then_succeed with patch("time.sleep"): mgr._async_queue.put(sess) assert retry_done.wait(timeout=10), "async writer never retried" mgr.shutdown() assert call_count[0] == 2 def _prime_migration_session(mgr, key, honcho_session_id, ai_peer_id="custom-ai"): """Cache a session whose user peer is what the REAL resolver returns for this manager — exactly what get_or_create stores — so the owner gate is tested against reachable states, not hand-picked peer ids.""" session = _make_session( key=key, user_peer_id=mgr._resolve_user_peer_id(key), assistant_peer_id=ai_peer_id, honcho_session_id=honcho_session_id, ) mgr._cache[session.key] = session honcho_session = MagicMock() mgr._sessions_cache[session.honcho_session_id] = honcho_session return session, honcho_session class TestMemoryFileMigrationTargets: def test_soul_upload_targets_ai_peer(self, tmp_path, make_manager): # peerName declares the owner; no runtime identity, so the session # resolves to the owner peer and migration proceeds. mgr = make_manager(write_frequency="turn", peer_name="custom-user") session, honcho_session = _prime_migration_session(mgr, "cli:test", "cli-test") assert session.user_peer_id == "custom-user" user_peer = MagicMock(name="user-peer") ai_peer = MagicMock(name="ai-peer") mgr._peers_cache[session.user_peer_id] = user_peer mgr._peers_cache[session.assistant_peer_id] = ai_peer (tmp_path / "MEMORY.md").write_text("memory facts", encoding="utf-8") (tmp_path / "USER.md").write_text("user profile", encoding="utf-8") (tmp_path / "SOUL.md").write_text("ai identity", encoding="utf-8") uploaded = mgr.migrate_memory_files(session.key, str(tmp_path)) assert uploaded is True assert honcho_session.upload_file.call_count == 3 peer_by_upload_name = {} for call_args in honcho_session.upload_file.call_args_list: payload = call_args.kwargs["file"] peer_by_upload_name[payload[0]] = call_args.kwargs["peer"] assert peer_by_upload_name["consolidated_memory.md"] is user_peer assert peer_by_upload_name["user_profile.md"] is user_peer assert peer_by_upload_name["agent_soul.md"] is ai_peer class TestMemoryFileMigrationOwnerGate: def test_non_owner_gateway_user_is_skipped(self, tmp_path, make_manager): """The shared-channel scenario: a declared owner exists, but the session was triggered by someone else's platform identity. The old gate (re-resolving the session's own peer) passed here.""" mgr = make_manager( write_frequency="turn", peer_name="owner-user", runtime_user_peer_name="some-other-human", ) session, honcho_session = _prime_migration_session( mgr, "discord:shared", "shared-chan" ) assert session.user_peer_id == "some-other-human" (tmp_path / "MEMORY.md").write_text("owner facts", encoding="utf-8") uploaded = mgr.migrate_memory_files(session.key, str(tmp_path)) assert uploaded is False assert honcho_session.upload_file.call_count == 0 def test_no_declared_owner_with_gateway_identity_is_skipped( self, tmp_path, make_manager): """Without peerName nobody messaging through a gateway can be proven to be the owner — migration must not run.""" mgr = make_manager( write_frequency="turn", runtime_user_peer_name="discord-123", ) session, honcho_session = _prime_migration_session( mgr, "discord:shared", "shared-chan" ) (tmp_path / "MEMORY.md").write_text("owner facts", encoding="utf-8") uploaded = mgr.migrate_memory_files(session.key, str(tmp_path)) assert uploaded is False assert honcho_session.upload_file.call_count == 0 def test_no_declared_owner_without_identity_has_no_session_to_migrate(self, tmp_path, make_manager): """No peerName and no runtime identity: the resolver refuses to name a peer (#93326), so no session exists for the owner gate and nothing is uploaded.""" mgr = make_manager(write_frequency="turn") (tmp_path / "MEMORY.md").write_text("memory facts", encoding="utf-8") with pytest.raises(HonchoPeerUnresolvedError): _prime_migration_session(mgr, "cli:test", "cli-test") assert mgr.migrate_memory_files("cli:test", str(tmp_path)) is False assert make_manager.client.session.return_value.upload_file.call_count == 0 def test_aliased_owner_identity_migrates(self, tmp_path, make_manager): """An alias mapping the owner's platform ID onto peerName makes that gateway identity the owner.""" mgr = make_manager( write_frequency="turn", peer_name="owner-user", user_peer_aliases={"discord-999": "owner-user"}, runtime_user_peer_name="discord-999", ) session, honcho_session = _prime_migration_session( mgr, "discord:dm", "discord-dm" ) assert session.user_peer_id == "owner-user" mgr._peers_cache[session.user_peer_id] = MagicMock() mgr._peers_cache[session.assistant_peer_id] = MagicMock() (tmp_path / "USER.md").write_text("user profile", encoding="utf-8") uploaded = mgr.migrate_memory_files(session.key, str(tmp_path)) assert uploaded is True assert honcho_session.upload_file.call_count == 1 def test_pinned_peer_name_migrates(self, tmp_path, make_manager): """pinPeerName collapses every identity onto the owner peer by explicit config, so the files land on the peer they describe.""" mgr = make_manager( write_frequency="turn", peer_name="owner-user", pin_peer_name=True, runtime_user_peer_name="anyone-at-all", ) session, honcho_session = _prime_migration_session( mgr, "discord:shared", "shared-chan" ) assert session.user_peer_id == "owner-user" mgr._peers_cache[session.user_peer_id] = MagicMock() mgr._peers_cache[session.assistant_peer_id] = MagicMock() (tmp_path / "MEMORY.md").write_text("memory facts", encoding="utf-8") uploaded = mgr.migrate_memory_files(session.key, str(tmp_path)) assert uploaded is True assert honcho_session.upload_file.call_count == 1 class TestPrefetchCacheAccessors: def test_set_and_pop_context_result(self, make_manager): mgr = make_manager(write_frequency="turn") payload = {"representation": "Known user", "card": "prefers concise replies"} mgr.set_context_result("cli:test", payload) 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 _blocking_remote(self, mgr, session): """Remote whose add_messages blocks until the returned release event is set.""" upload_started, release_upload = threading.Event(), threading.Event() def blocking_add_messages(_messages): upload_started.set() release_upload.wait(timeout=2) return self._wire_remote(mgr, session, blocking_add_messages), upload_started, release_upload 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") remote, upload_started, release_upload = self._blocking_remote(mgr, session) 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 remote, upload_started, release_upload = self._blocking_remote(mgr, session) 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