diff --git a/hermes_cli/oneshot.py b/hermes_cli/oneshot.py index 8aa6c9662d4ce..59076c677ed9c 100644 --- a/hermes_cli/oneshot.py +++ b/hermes_cli/oneshot.py @@ -319,6 +319,47 @@ def _create_session_db_for_oneshot(): return None +def _shutdown_memory_provider_with_transcript(agent) -> None: + """Shut down ``agent``'s memory provider, forwarding the real transcript. + + Mirrors cli.py:_run_cleanup: forward the agent's own ``_session_messages`` + so memory providers' ``on_session_end`` hooks see the real conversation + instead of an empty list. ``_session_messages`` is set on + ``AIAgent.__init__`` and refreshed every turn via ``_persist_session``. + Fall back to the no-messages call on test stubs / partially-initialised + agents where the attribute is missing or not a list. Never raises. + """ + if agent is None or not hasattr(agent, "shutdown_memory_provider"): + return + try: + session_messages = getattr(agent, "_session_messages", None) + if isinstance(session_messages, list): + agent.shutdown_memory_provider(session_messages) + else: + agent.shutdown_memory_provider() + except Exception: + logging.debug("oneshot memory/context cleanup failed", exc_info=True) + + +def _make_oneshot_memory_shutdown(get_agent): + """Build the once-only memory-shutdown hook for a oneshot run. + + The returned callable is shared by the explicit ``finally`` cleanup and + the atexit fallback in ``_run_agent`` — whichever fires first wins, the + other becomes a no-op. The agent is behind a callable because the hook + is registered before ``AIAgent(...)`` finishes constructing. + """ + done = [False] + + def _shutdown_once() -> None: + if done[0]: + return + done[0] = True + _shutdown_memory_provider_with_transcript(get_agent()) + + return _shutdown_once + + def _run_agent( prompt: str, model: Optional[str] = None, @@ -424,6 +465,22 @@ def _run_agent( # raises on a provider/config error. The one-shot exit path hard-exits via # os._exit and skips finalizers, so an un-closed connection here would leak. agent = None + + # Oneshot bypasses the interactive CLI's atexit/_run_cleanup wiring, so + # memory providers (e.g. Honcho) whose daemon worker threads are still + # blocked in HTTP recv at interpreter exit never get shutdown() called. + # CPython then abandons those threads at Py_FinalizeEx and tears the + # interpreter down around them, producing SIGABRT (exit 134) with no + # Python traceback. The ``finally`` below handles the normal path; + # register the same once-guarded hook as an atexit fallback for exit + # paths that skip it — whichever fires first wins, the other becomes a + # no-op. Both paths forward the real transcript (see + # _shutdown_memory_provider_with_transcript). + import atexit as _atexit + + _shutdown_oneshot_memory_provider = _make_oneshot_memory_shutdown(lambda: agent) + _atexit.register(_shutdown_oneshot_memory_provider) + try: # Read the effective fallback chain from profile config so oneshot # workers honour the same merge semantics as interactive CLI and @@ -470,14 +527,7 @@ def _run_agent( # NOT cli.py:_run_cleanup — oneshot has no _active_agent_ref and must # close the agent explicitly because the hard-exit path skips finalizers. if agent is not None: - try: - session_messages = getattr(agent, "_session_messages", None) - if isinstance(session_messages, list): - agent.shutdown_memory_provider(session_messages) - else: - agent.shutdown_memory_provider() - except Exception: - logging.debug("oneshot memory/context cleanup failed", exc_info=True) + _shutdown_oneshot_memory_provider() try: agent.close() except Exception: diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index ea24ed064fc7f..4b62a20885ede 100644 --- a/plugins/memory/honcho/__init__.py +++ b/plugins/memory/honcho/__init__.py @@ -1607,16 +1607,83 @@ def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str: return tool_error(f"Honcho {tool_name} failed: {e}") def shutdown(self) -> None: + # CPython aborts (SIGABRT, exit 134) when daemon threads are still + # blocked in C-level I/O (httpx socket recv) at Py_FinalizeEx. + # The precise internal path is not fully characterized — likely + # Py_FatalError("Invalid thread state") or a glibc assert in + # pthread_mutex_destroy when a daemon thread still holds a lock + # during module/state teardown. CPython does not forcibly kill + # daemon threads; it abandons them and tears down the interpreter + # out from under them. See CPython gh-97940 / bpo-20526. + # + # Closing the Honcho SDK's underlying httpx.Client is the key step: + # it interrupts any worker thread blocked in sock_recv (httpx raises + # a connection-closed error that the worker's try/except absorbs), + # so the subsequent joins actually complete instead of timing out + # against a 30s read poll. + # + # Worst-case shutdown latency: ~2s (init join) + 2×2s (prefetch/sync) + # + manager shutdown (10s async join + 5s/prefetch) + 3×5s re-join + # ≈ 30s+ if anything is wedged. Acceptable vs. a crash. + if self._init_thread and self._init_thread.is_alive(): + self._init_thread.join(timeout=2.0) + for t in (self._prefetch_thread, self._sync_thread): if t and t.is_alive(): - t.join(timeout=5.0) - # Flush any remaining messages - if self._manager and not (self._init_thread and self._init_thread.is_alive() and not self._session_initialized): + t.join(timeout=2.0) + + if self._manager: + if not (self._init_thread and self._init_thread.is_alive()): + try: + self._manager.flush_all() + except Exception: + pass + # Close the Honcho httpx.Client FIRST so threads blocked in + # socket recv are interrupted — otherwise the manager's shutdown() + # joins time out against the 30s read poll. + self._close_honcho_http_client() + + # Now join the manager's async writer + prefetch threads (they + # should unblock quickly after the client close). try: - self._manager.flush_all() + self._manager.shutdown() except Exception: pass + # Re-join tracked threads after the http client close — they may + # have been blocked in socket recv during the first join attempt. + for t in (self._init_thread, self._prefetch_thread, self._sync_thread): + if t and t.is_alive(): + t.join(timeout=5.0) + + def _close_honcho_http_client(self) -> None: + """Close the Honcho SDK's httpx.Client to interrupt in-flight recvs. + + The honcho-ai SDK stores its sync HTTP client on ``Honcho._http`` + (a ``HonchoHTTPClient`` wrapping ``httpx.Client``). Closing it + forces any worker thread blocked in a socket recv to error out, + which is required for a clean interpreter shutdown. + """ + client = getattr(self, "_manager", None) + client = getattr(client, "_honcho", None) if client else None + if client is None: + logger.warning( + "shutdown_memory_provider: could not locate Honcho SDK client " + "(_manager._honcho) — httpx client not closed, daemon threads " + "may linger. SDK attribute path may have drifted." + ) + return + for attr in ("_http", "_async_http"): + http_client = getattr(client, attr, None) + if http_client is None: + continue + closer = getattr(http_client, "close", None) + if callable(closer): + try: + closer() + except Exception: + pass + # --------------------------------------------------------------------------- # Plugin entry point diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index b8d2db4ef3c90..57eba9982d74d 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -202,6 +202,12 @@ def __init__( self._async_queue: queue.Queue | None = None self._async_thread: threading.Thread | None = None self._async_thread_lock = threading.Lock() + # Tracked fire-and-forget context-prefetch threads, joined in shutdown() + # so none is left blocked in HTTP recv at interpreter teardown (which + # aborts CPython — see HonchoMemoryProvider.shutdown). + self._context_prefetch_threads: list[threading.Thread] = [] + self._prefetch_threads_lock = threading.Lock() + self._closed = False if write_frequency == "async": self._async_queue = queue.Queue() @@ -772,12 +778,41 @@ def _ensure_async_writer(self) -> None: self._async_thread.start() def shutdown(self) -> None: - """Gracefully shut down the async writer thread.""" + """Gracefully shut down background worker threads. + + Joins the async writer and any in-flight context-prefetch threads so + none is left blocked in HTTP recv at interpreter teardown (which + aborts CPython during Py_FinalizeEx — daemon threads are abandoned, + not killed, and the interpreter tears down around them; gh-97940). + """ + # Atomically mark closed AND snapshot the tracked prefetch threads + # under the registration lock. A concurrent prefetch_context() holds + # the same lock across its closed-check/register/start sequence, so + # it either completes registration before this snapshot (and its + # thread is joined below) or observes _closed afterwards and never + # starts a thread. Without the atomicity a prefetch could pass the + # closed-check, then register+start after the snapshot — a lost + # thread left blocked in HTTP recv at teardown (TOCTOU → SIGABRT). + with self._prefetch_threads_lock: + if self._closed: + return + self._closed = True + prefetch_snapshot = list(self._context_prefetch_threads) if self._async_queue is not None: self.flush_all() 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) + # Join context-prefetch threads, but keep references to any that + # don't join within the timeout so a later shutdown phase can retry. + alive = [] + for t in prefetch_snapshot: + if t.is_alive(): + t.join(timeout=5.0) + if t.is_alive(): + alive.append(t) + with self._prefetch_threads_lock: + self._context_prefetch_threads = alive def delete(self, key: str) -> bool: """Delete a session from local cache.""" @@ -906,19 +941,36 @@ def _chat_once() -> str: return "" def prefetch_context(self, session_key: str, user_message: str | None = None) -> None: - """ - Fire get_prefetch_context in a background thread, caching the result. + """Fire get_prefetch_context in a background thread, caching the result. Non-blocking. Consumed next turn via pop_context_result(). This avoids a synchronous HTTP round-trip blocking every response. """ + if self._closed: + return # fast path; the authoritative check is under the lock below + def _run(): result = self.get_prefetch_context(session_key, user_message) if result: self.set_context_result(session_key, result) t = threading.Thread(target=_run, name="honcho-context-prefetch", daemon=True) - t.start() + # Check-closed, register, and start atomically under the same lock + # shutdown() uses to mark closed + snapshot the thread list: either + # this thread makes it into shutdown()'s snapshot (and is joined + # there), or _closed is already set and the thread is never started. + # Also prune completed threads to avoid unbounded list growth in + # long-lived sessions (e.g. gateway runs for days/weeks with many + # turns). Thread.start() is cheap and _run never touches this lock, + # so starting inside the critical section cannot deadlock. + with self._prefetch_threads_lock: + if self._closed: + return + self._context_prefetch_threads = [ + th for th in self._context_prefetch_threads if th.is_alive() + ] + self._context_prefetch_threads.append(t) + t.start() def set_context_result(self, session_key: str, result: dict[str, str]) -> None: """Store a prefetched context result in a thread-safe way.""" diff --git a/tests/hermes_cli/test_oneshot_memory_cleanup.py b/tests/hermes_cli/test_oneshot_memory_cleanup.py new file mode 100644 index 0000000000000..dce15bf03a616 --- /dev/null +++ b/tests/hermes_cli/test_oneshot_memory_cleanup.py @@ -0,0 +1,134 @@ +"""Tests for oneshot memory-provider cleanup (SIGABRT fix, #61875). + +Verifies that: +1. The explicit cleanup path forwards the agent's real transcript to + shutdown_memory_provider (mirroring cli.py:_run_cleanup) instead of + letting run_agent.py's on_session_end see an empty list. +2. Cleanup runs exactly once even when both the explicit ``finally`` path + and the atexit fallback fire. +3. Missing / non-list ``_session_messages`` falls back to the no-messages + call instead of raising. +""" + +from unittest.mock import patch + +from hermes_cli.oneshot import ( + _make_oneshot_memory_shutdown, + _run_agent, + _shutdown_memory_provider_with_transcript, +) + + +class AgentStub: + """Minimal stand-in for AIAgent recording memory-shutdown calls.""" + + _NO_MESSAGES = object() # sentinel: shutdown called without a transcript + + def __init__(self, messages=None): + if messages is not None: + self._session_messages = messages + self.shutdown_calls = [] + self.close_calls = 0 + + def run_conversation(self, prompt): + return {"final_response": "ok"} + + def shutdown_memory_provider(self, messages=_NO_MESSAGES): + self.shutdown_calls.append(messages) + + def close(self): + self.close_calls += 1 + + +def _drive_run_agent(agent): + """Run _run_agent with all heavy dependencies stubbed out. + + Returns ``(response, result, atexit_hook)`` where ``atexit_hook`` is the + fallback _run_agent registered (captured instead of installed, so tests + don't leak hooks into the interpreter). + """ + hooks = [] + with ( + patch("hermes_cli.config.load_config", return_value={}), + patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value={}), + patch("hermes_cli.tools_config._get_platform_tools", return_value=set()), + patch("run_agent.AIAgent", return_value=agent), + patch("hermes_cli.oneshot._create_session_db_for_oneshot", return_value=None), + patch("hermes_cli.oneshot.get_fallback_chain", return_value=None), + patch("atexit.register", side_effect=hooks.append), + ): + response, result = _run_agent("hi") + assert len(hooks) == 1, "oneshot must register exactly one atexit fallback" + return response, result, hooks[0] + + +class TestOneshotForwardsTranscript: + def test_explicit_cleanup_forwards_real_transcript(self, monkeypatch): + monkeypatch.delenv("HERMES_INFERENCE_MODEL", raising=False) + transcript = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "ok"}, + ] + agent = AgentStub(messages=transcript) + + response, _, _ = _drive_run_agent(agent) + + assert response == "ok" + assert agent.shutdown_calls == [transcript], ( + "explicit cleanup must forward agent._session_messages, " + "not call shutdown_memory_provider() bare" + ) + assert agent.close_calls == 1 + + def test_cleanup_runs_once_when_atexit_also_fires(self, monkeypatch): + monkeypatch.delenv("HERMES_INFERENCE_MODEL", raising=False) + agent = AgentStub(messages=[{"role": "user", "content": "hi"}]) + + _, _, atexit_hook = _drive_run_agent(agent) + assert len(agent.shutdown_calls) == 1 # explicit finally path fired + + atexit_hook() # simulated interpreter-exit fallback + atexit_hook() # atexit must also tolerate repeat invocation + + assert len(agent.shutdown_calls) == 1, ( + "atexit fallback must be a no-op after the explicit cleanup ran" + ) + + def test_atexit_fallback_forwards_transcript_when_it_fires_first(self): + """When only the atexit path runs (finally skipped, e.g. hard-exit + wedge), the fallback itself must forward the real transcript.""" + transcript = [{"role": "user", "content": "hi"}] + agent = AgentStub(messages=transcript) + hook = _make_oneshot_memory_shutdown(lambda: agent) + + hook() + + assert agent.shutdown_calls == [transcript] + + def test_missing_session_messages_falls_back_to_bare_call(self): + agent = AgentStub(messages=None) # attribute absent entirely + _shutdown_memory_provider_with_transcript(agent) + assert agent.shutdown_calls == [AgentStub._NO_MESSAGES] + + def test_non_list_session_messages_falls_back_to_bare_call(self): + agent = AgentStub(messages=None) + agent._session_messages = "not-a-list" + _shutdown_memory_provider_with_transcript(agent) + assert agent.shutdown_calls == [AgentStub._NO_MESSAGES] + + def test_none_agent_is_noop(self): + _shutdown_memory_provider_with_transcript(None) # must not raise + + def test_shutdown_exceptions_are_swallowed(self): + class ExplodingAgent(AgentStub): + def shutdown_memory_provider(self, messages=AgentStub._NO_MESSAGES): + raise RuntimeError("boom") + + _shutdown_memory_provider_with_transcript(ExplodingAgent(messages=[])) + + def test_make_oneshot_memory_shutdown_is_idempotent(self): + agent = AgentStub(messages=[]) + hook = _make_oneshot_memory_shutdown(lambda: agent) + hook() + hook() + assert len(agent.shutdown_calls) == 1 diff --git a/tests/honcho_plugin/test_honcho_shutdown.py b/tests/honcho_plugin/test_honcho_shutdown.py new file mode 100644 index 0000000000000..96698d322cb1e --- /dev/null +++ b/tests/honcho_plugin/test_honcho_shutdown.py @@ -0,0 +1,219 @@ +"""Tests for Honcho memory provider shutdown — SIGABRT prevention. + +Verifies that: +1. The prefetch-thread TOCTOU race is guarded by _prefetch_threads_lock. +2. _close_honcho_http_client warns when the Honcho SDK client can't be + located (silent SDK attribute drift detection). +3. shutdown() joins all tracked prefetch threads. +""" + +import logging +import threading +from unittest.mock import MagicMock, patch + +import pytest + +from plugins.memory.honcho import HonchoMemoryProvider + + +class TestPrefetchThreadLock: + """Verify the _prefetch_threads_lock guards concurrent access.""" + + def test_lock_exists(self): + """HonchoSession must have a _prefetch_threads_lock attribute.""" + from plugins.memory.honcho.session import HonchoSessionManager + + manager = HonchoSessionManager.__new__(HonchoSessionManager) + # Verify the attribute would exist on a real instance + assert hasattr(HonchoSessionManager, "__init__") + # Check the source includes the lock + import inspect + + src = inspect.getsource(HonchoSessionManager.__init__) + assert "_prefetch_threads_lock" in src, ( + "HonchoSessionManager.__init__ must initialize _prefetch_threads_lock" + ) + + def test_prefetch_context_uses_lock(self): + """prefetch_context must acquire _prefetch_threads_lock.""" + import inspect + + from plugins.memory.honcho.session import HonchoSessionManager + + src = inspect.getsource(HonchoSessionManager.prefetch_context) + assert "_prefetch_threads_lock" in src, ( + "prefetch_context must use _prefetch_threads_lock to guard the thread list" + ) + + def test_shutdown_uses_lock(self): + """shutdown must acquire _prefetch_threads_lock for snapshot.""" + import inspect + + from plugins.memory.honcho.session import HonchoSessionManager + + src = inspect.getsource(HonchoSessionManager.shutdown) + assert "_prefetch_threads_lock" in src, ( + "shutdown must use _prefetch_threads_lock to snapshot prefetch threads" + ) + + +class TestCloseHonchoHttpClientWarnings: + """Verify SDK drift is observable via logger.warning.""" + + def test_warns_when_manager_missing(self, caplog): + """When _manager is None, a warning must be logged.""" + provider = HonchoMemoryProvider.__new__(HonchoMemoryProvider) + provider._manager = None + + with caplog.at_level(logging.WARNING): + provider._close_honcho_http_client() + + assert any("could not locate Honcho SDK client" in r.message for r in caplog.records), ( + "Must warn when Honcho SDK client can't be found" + ) + + def test_warns_when_honcho_attr_missing(self, caplog): + """When _manager._honcho is missing, a warning must be logged.""" + provider = HonchoMemoryProvider.__new__(HonchoMemoryProvider) + mock_manager = MagicMock() + del mock_manager._honcho # Make getattr return a spec'd mock, not None + provider._manager = mock_manager + + # Actually we need _honcho to be None + mock_manager._honcho = None + provider._manager = mock_manager + + with caplog.at_level(logging.WARNING): + provider._close_honcho_http_client() + + assert any("could not locate Honcho SDK client" in r.message for r in caplog.records), ( + "Must warn when _manager._honcho is None" + ) + + def test_no_warning_when_client_found(self, caplog): + """When the client path is valid, no drift warning is logged.""" + provider = HonchoMemoryProvider.__new__(HonchoMemoryProvider) + mock_http = MagicMock() + mock_http.close = MagicMock() + mock_honcho = MagicMock() + mock_honcho._http = mock_http + mock_honcho._async_http = None + mock_manager = MagicMock() + mock_manager._honcho = mock_honcho + provider._manager = mock_manager + + with caplog.at_level(logging.WARNING): + provider._close_honcho_http_client() + + assert not any("could not locate" in r.message for r in caplog.records), ( + "Should not warn when client is found" + ) + mock_http.close.assert_called_once() + + +def _bare_manager(): + """A HonchoSessionManager with just the shutdown/prefetch plumbing.""" + from plugins.memory.honcho.session import HonchoSessionManager + + manager = HonchoSessionManager.__new__(HonchoSessionManager) + manager._closed = False + manager._prefetch_threads_lock = threading.Lock() + manager._context_prefetch_threads = [] + manager._async_queue = None + manager._async_thread = None + return manager + + +class TestShutdownPrefetchRace: + """shutdown() vs prefetch_context() — no thread may start after close.""" + + def test_prefetch_after_shutdown_is_noop(self): + """A prefetch attempted after shutdown must not start or track a thread.""" + manager = _bare_manager() + manager.shutdown() + + with patch.object(threading.Thread, "start") as mock_start: + manager.prefetch_context("key") + + mock_start.assert_not_called() + assert manager._context_prefetch_threads == [] + + def test_shutdown_interleaved_mid_prefetch_blocks_registration(self): + """Deterministic TOCTOU: shutdown() lands between prefetch_context()'s + fast closed-check and its registration. The lock-guarded re-check must + prevent the thread from being started or tracked.""" + from plugins.memory.honcho import session as session_mod + + manager = _bare_manager() + created: list[threading.Thread] = [] + real_thread = threading.Thread + + def racing_thread(*args, **kwargs): + # Fires after prefetch_context() has already passed its fast + # `if self._closed` check — the exact TOCTOU window. + manager.shutdown() + t = real_thread(*args, **kwargs) + created.append(t) + return t + + with patch.object(session_mod.threading, "Thread", side_effect=racing_thread): + manager.prefetch_context("key") + + assert manager._closed is True + assert created, "prefetch_context should have constructed its thread" + assert created[0].ident is None, ( + "prefetch thread must never be started once shutdown() has run" + ) + assert manager._context_prefetch_threads == [], ( + "prefetch thread must not be tracked after shutdown()'s snapshot" + ) + + def test_prefetch_before_shutdown_lands_in_snapshot(self): + """A prefetch registered before shutdown() must be joined by it.""" + manager = _bare_manager() + release = threading.Event() + # Instance-level stubs: keep the worker alive until released so the + # thread is still running when shutdown() takes its snapshot. + manager.get_prefetch_context = lambda *a, **k: release.wait(timeout=5) and None + manager.set_context_result = lambda *a, **k: None + + manager.prefetch_context("key") + assert len(manager._context_prefetch_threads) == 1 + worker = manager._context_prefetch_threads[0] + release.set() + manager.shutdown() + + assert not worker.is_alive(), "shutdown() must join tracked prefetch threads" + + def test_shutdown_is_idempotent(self): + """A second shutdown() must return early without re-joining.""" + manager = _bare_manager() + manager.shutdown() + assert manager._closed is True + manager.shutdown() # must not raise or deadlock + assert manager._closed is True + + +class TestShutdownJoinPrefetchThreads: + """Verify shutdown actually joins prefetch threads.""" + + def test_shutdown_joins_tracked_thread(self): + """A thread in _context_prefetch_threads must be joined during shutdown.""" + from plugins.memory.honcho.session import HonchoSessionManager + + manager = HonchoSessionManager.__new__(HonchoSessionManager) + manager._closed = False + manager._prefetch_threads_lock = threading.Lock() + manager._context_prefetch_threads = [] + manager._async_queue = None + manager._async_thread = None + + # Track a fake completed thread + fake_thread = MagicMock() + fake_thread.is_alive.return_value = False + manager._context_prefetch_threads = [fake_thread] + + manager.shutdown() + + # Thread was checked + fake_thread.is_alive.assert_called()