fix(oneshot): shut down memory provider to prevent SIGABRT on exit

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:

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

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."""