diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index 97bdb3b35ee..d4a53a6e424 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -140,17 +140,16 @@ class HonchoSessionManager: config.dialectic_max_input_chars if config else 10000 ) - # Async write queue — started lazily on first enqueue + # Async write queue — the writer thread starts lazily on first enqueue + # (see _ensure_async_writer). Constructing a manager must not spawn + # background work or touch the network: unit tests build managers with + # mocked clients, and an eagerly-started writer raced ahead of the mock + # and wrote test messages to a live local Honcho. self._async_queue: queue.Queue | None = None self._async_thread: threading.Thread | None = None + self._async_thread_lock = threading.Lock() if write_frequency == "async": self._async_queue = queue.Queue() - self._async_thread = threading.Thread( - target=self._async_writer_loop, - name="honcho-async-writer", - daemon=True, - ) - self._async_thread.start() @property def honcho(self) -> Honcho: @@ -511,6 +510,7 @@ class HonchoSessionManager: if wf == "async": if self._async_queue is not None: + self._ensure_async_writer() self._async_queue.put(session) elif wf == "turn": self._flush_session(session) @@ -545,12 +545,26 @@ class HonchoSessionManager: except queue.Empty: break + def _ensure_async_writer(self) -> None: + """Start the async writer on first enqueue (idempotent, thread-safe).""" + if self._async_thread is not None and self._async_thread.is_alive(): + return + with self._async_thread_lock: + if self._async_thread is None or not self._async_thread.is_alive(): + self._async_thread = threading.Thread( + target=self._async_writer_loop, + name="honcho-async-writer", + daemon=True, + ) + self._async_thread.start() + def shutdown(self) -> None: """Gracefully shut down the async writer thread.""" - if self._async_queue is not None and self._async_thread is not None: + if self._async_queue is not None: self.flush_all() - self._async_queue.put(_ASYNC_SHUTDOWN) - self._async_thread.join(timeout=10) + if self._async_thread is not None and self._async_thread.is_alive(): + self._async_queue.put(_ASYNC_SHUTDOWN) + self._async_thread.join(timeout=10) def delete(self, key: str) -> bool: """Delete a session from local cache.""" diff --git a/tests/honcho_plugin/conftest.py b/tests/honcho_plugin/conftest.py new file mode 100644 index 00000000000..7667a46c6f3 --- /dev/null +++ b/tests/honcho_plugin/conftest.py @@ -0,0 +1,77 @@ +"""Package-level isolation for Honcho unit tests. + +Contract (B8): unit tests in this package make ZERO network requests, even +when a live local Honcho is reachable and ambient production config exists. +A real incident proved unique fixtures are not enough - test messages were +written into the production workspace. The guard below turns any connection +attempt into a hard failure; individually marked tests may opt out with +@pytest.mark.allow_network. +""" + +import socket +import threading + +import pytest + + +def pytest_configure(config): + config.addinivalue_line( + "markers", + "allow_network: opt-in marker for tests that intentionally touch the network", + ) + config.addinivalue_line( + "markers", + "expect_network_attempts: test asserts on blocked attempts itself", + ) + + +@pytest.fixture +def network_attempts(): + """Recorded connection attempts (visible to tests for assertions).""" + return [] + + +@pytest.fixture(autouse=True) +def _no_network(request, monkeypatch, network_attempts): + """Fail any test in this package that attempts a real socket connection. + + Recording + teardown assert (not just raising) catches attempts even when + intermediate code swallows the exception (the async writer retries and + logs errors instead of propagating them). + """ + if request.node.get_closest_marker("allow_network"): + yield + return + + def _blocked_connect(self, address, *args, **kwargs): + network_attempts.append(address) + raise RuntimeError(f"network disabled in honcho unit tests: {address!r}") + + def _blocked_create_connection(address, *args, **kwargs): + network_attempts.append(address) + raise RuntimeError(f"network disabled in honcho unit tests: {address!r}") + + monkeypatch.setattr(socket.socket, "connect", _blocked_connect) + monkeypatch.setattr(socket, "create_connection", _blocked_create_connection) + yield + leftover = list(network_attempts) + if request.node.get_closest_marker("expect_network_attempts"): + return + assert not leftover, ( + f"unit test attempted network connections: {leftover} - " + "inject a fake client/factory before constructing the manager" + ) + + +@pytest.fixture(autouse=True) +def _no_leaked_writer_threads(): + """Every async manager must be shut down by the test that created it.""" + yield + leaked = [ + t for t in threading.enumerate() + if t.name == "honcho-async-writer" and t.is_alive() + ] + assert not leaked, ( + f"leaked honcho-async-writer threads: {len(leaked)} - " + "call manager.shutdown() (use the make_manager fixture)" + ) diff --git a/tests/honcho_plugin/test_async_memory.py b/tests/honcho_plugin/test_async_memory.py index d217b87e038..a6dddf4c292 100644 --- a/tests/honcho_plugin/test_async_memory.py +++ b/tests/honcho_plugin/test_async_memory.py @@ -13,6 +13,8 @@ import json import threading from unittest.mock import MagicMock, patch +import pytest + from plugins.memory.honcho.client import HonchoClientConfig from plugins.memory.honcho.session import ( @@ -35,15 +37,37 @@ def _make_session(**kwargs) -> HonchoSession: ) -def _make_manager(write_frequency="turn") -> HonchoSessionManager: - cfg = HonchoClientConfig( - write_frequency=write_frequency, - api_key="test-key", - enabled=True, - ) - mgr = HonchoSessionManager(config=cfg) - mgr._honcho = MagicMock() - return mgr +# 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") -> HonchoSessionManager: + cfg = HonchoClientConfig( + write_frequency=write_frequency, + api_key="test-key", + enabled=True, + ) + mgr = HonchoSessionManager(honcho=client, config=cfg) + created.append(mgr) + return mgr + + _make.client = client + yield _make + for mgr in created: + mgr.shutdown() # --------------------------------------------------------------------------- @@ -146,22 +170,22 @@ class TestSaveRouting: mgr._cache[sess.key] = sess return sess - def test_turn_flushes_immediately(self): - mgr = _make_manager(write_frequency="turn") + 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): - mgr = _make_manager(write_frequency="session") + 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): - mgr = _make_manager(write_frequency="async") + 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) @@ -169,8 +193,8 @@ class TestSaveRouting: mock_flush.assert_not_called() assert not mgr._async_queue.empty() - def test_int_frequency_flushes_on_nth_turn(self): - mgr = _make_manager(write_frequency=3) + 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 @@ -179,14 +203,24 @@ class TestSaveRouting: mgr.save(sess) # turn 3 assert mock_flush.call_count == 1 + def test_int_frequency_skips_other_turns(self, make_manager): + mgr = make_manager(write_frequency=5) + sess = self._make_session_with_message(mgr) + with patch.object(mgr, "_flush_session") as mock_flush: + for _ in range(4): + mgr.save(sess) + assert mock_flush.call_count == 0 + mgr.save(sess) # turn 5 + assert mock_flush.call_count == 1 + # --------------------------------------------------------------------------- # flush_all() # --------------------------------------------------------------------------- class TestFlushAll: - def test_flushes_all_cached_sessions(self): - mgr = _make_manager(write_frequency="session") + 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") @@ -197,8 +231,8 @@ class TestFlushAll: mgr.flush_all() assert mock_flush.call_count == 2 - def test_flush_all_drains_async_queue(self): - mgr = _make_manager(write_frequency="async") + def test_flush_all_drains_async_queue(self, make_manager): + mgr = make_manager(write_frequency="async") sess = _make_session() sess.add_message("user", "pending") @@ -211,37 +245,86 @@ class TestFlushAll: # 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_started_on_async_mode(self): - mgr = _make_manager(write_frequency="async") + 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): - mgr = _make_manager(write_frequency="turn") + 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): - mgr = _make_manager(write_frequency="async") + def test_shutdown_joins_thread(self, make_manager): + mgr = make_manager(write_frequency="async") + mgr._ensure_async_writer() 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() + 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_sentinel_stops_loop(self, make_manager): + mgr = make_manager(write_frequency="async") + mgr._ensure_async_writer() + thread = mgr._async_thread + mgr.shutdown() + thread.join(timeout=10) + assert not thread.is_alive() + + 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 + # --------------------------------------------------------------------------- # async retry on failure # --------------------------------------------------------------------------- class TestAsyncWriterRetry: - def test_retries_once_on_failure(self): - mgr = _make_manager(write_frequency="async") + def test_retries_once_on_failure(self, make_manager): + mgr = make_manager(write_frequency="async") + mgr._ensure_async_writer() sess = _make_session() sess.add_message("user", "msg") @@ -264,8 +347,9 @@ class TestAsyncWriterRetry: mgr.shutdown() assert call_count[0] == 2 - def test_drops_after_two_failures(self): - mgr = _make_manager(write_frequency="async") + def test_drops_after_two_failures(self, make_manager): + mgr = make_manager(write_frequency="async") + mgr._ensure_async_writer() sess = _make_session() sess.add_message("user", "msg") @@ -289,10 +373,34 @@ class TestAsyncWriterRetry: 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() + 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 + class TestMemoryFileMigrationTargets: - def test_soul_upload_targets_ai_peer(self, tmp_path): - mgr = _make_manager(write_frequency="turn") + def test_soul_upload_targets_ai_peer(self, tmp_path, make_manager): + mgr = make_manager(write_frequency="turn") session = _make_session( key="cli:test", user_peer_id="custom-user", @@ -339,8 +447,8 @@ class TestNewConfigFieldDefaults: class TestPrefetchCacheAccessors: - def test_set_and_pop_context_result(self): - mgr = _make_manager(write_frequency="turn") + 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) diff --git a/tests/honcho_plugin/test_network_isolation.py b/tests/honcho_plugin/test_network_isolation.py new file mode 100644 index 00000000000..f59261b9d0d --- /dev/null +++ b/tests/honcho_plugin/test_network_isolation.py @@ -0,0 +1,108 @@ +"""B8 regressions: honcho unit tests must never touch the network. + +Real incident: an async manager built with a late MagicMock swap wrote test +messages (session cli-test, hello/hi) into the production workspace of a live +local Honcho. These tests pin the isolation contract. +""" + +import json +import socket +import threading +import time +from unittest.mock import MagicMock + +import pytest + +from plugins.memory.honcho import session as session_module +from plugins.memory.honcho.client import HonchoClientConfig +from plugins.memory.honcho.session import HonchoSession, HonchoSessionManager + + +def _session(**kw) -> HonchoSession: + return HonchoSession( + key=kw.get("key", "cli:isolation"), + user_peer_id="eri", + assistant_peer_id="hermes", + honcho_session_id=kw.get("sid", "cli-isolation"), + messages=kw.get("messages", []), + ) + + +class TestConstructorRace: + def test_fake_factory_before_constructor_gets_all_calls(self, monkeypatch): + """Regression 1: fake factory installed BEFORE the constructor receives + every flush; no real transport is ever touched (network guard active).""" + fake = MagicMock() + monkeypatch.setattr(session_module, "get_honcho_client", lambda *a, **k: fake) + cfg = HonchoClientConfig(write_frequency="async", api_key="test-key", enabled=True) + mgr = HonchoSessionManager(honcho=fake, config=cfg) + try: + sess = _session() + sess.add_message("user", "hello") + sess.add_message("assistant", "hi") + mgr.save(sess) + deadline = time.time() + 3.0 + while not fake.method_calls and time.time() < deadline: + time.sleep(0.05) + assert fake.method_calls, "fake client never received the flush" + finally: + mgr.shutdown() + + def test_construction_spawns_no_thread_and_no_network(self): + """Constructing a manager (async mode) does nothing externally.""" + cfg = HonchoClientConfig(write_frequency="async", api_key="test-key", enabled=True) + mgr = HonchoSessionManager(config=cfg) + try: + assert mgr._async_thread is None + finally: + mgr.shutdown() + + +class TestAmbientProductionConfig: + def test_ambient_live_config_produces_zero_requests(self, tmp_path, monkeypatch): + """Regression 2: ambient HERMES_HOME with a live URL must not leak + requests - hygiene (factory injection) keeps the suite green.""" + home = tmp_path / "hermes-home" + home.mkdir() + (home / "honcho.json").write_text(json.dumps({ + "baseUrl": "http://localhost:8000", + "workspace": "iris_curated_v1", + "hosts": {"hermes": {"apiKey": "live-looking-key", "saveMessages": True}}, + })) + monkeypatch.setenv("HERMES_HOME", str(home)) + fake = MagicMock() + monkeypatch.setattr(session_module, "get_honcho_client", lambda *a, **k: fake) + cfg = HonchoClientConfig(write_frequency="async", api_key="live-looking-key", enabled=True) + mgr = HonchoSessionManager(honcho=fake, config=cfg) + try: + sess = _session(sid="cli-test") + sess.add_message("user", "hello") + mgr.save(sess) + mgr.flush_all() + finally: + mgr.shutdown() + # teardown-assert конфтеста дополнительно проверит network_attempts == [] + + @pytest.mark.expect_network_attempts + def test_guard_blocks_real_connections(self, network_attempts): + """The guard itself works: a raw connection attempt is recorded+raised.""" + with pytest.raises(RuntimeError, match="network disabled"): + socket.create_connection(("127.0.0.1", 8000), timeout=1) + assert network_attempts == [("127.0.0.1", 8000)] + + +class TestThreadCleanup: + def test_shutdown_leaves_no_writer_threads(self, monkeypatch): + """Regression 3: after shutdown - queue drained, thread stopped.""" + fake = MagicMock() + monkeypatch.setattr(session_module, "get_honcho_client", lambda *a, **k: fake) + cfg = HonchoClientConfig(write_frequency="async", api_key="test-key", enabled=True) + mgr = HonchoSessionManager(honcho=fake, config=cfg) + sess = _session() + sess.add_message("user", "bye") + mgr.save(sess) + mgr.shutdown() + assert mgr._async_queue.empty() + assert not any( + t.name == "honcho-async-writer" and t.is_alive() for t in threading.enumerate() + ) diff --git a/tests/honcho_plugin/test_oauth_flow.py b/tests/honcho_plugin/test_oauth_flow.py index 4b4d8dea9ed..f590932e022 100644 --- a/tests/honcho_plugin/test_oauth_flow.py +++ b/tests/honcho_plugin/test_oauth_flow.py @@ -18,6 +18,10 @@ import pytest from plugins.memory.honcho import oauth, oauth_flow +# Opt-in: эти тесты поднимают собственный loopback OAuth-сервер внутри теста +# и коннектятся к нему же - внешней сети нет (guard B8 это разрешает явно). +pytestmark = pytest.mark.allow_network + class _FakeAS(BaseHTTPRequestHandler): """Minimal OAuth 2.1 AS: /authorize 302s to the callback; /oauth/token mints."""