fix(honcho): drain context prefetch threads during shutdown

plugins/memory/honcho/__init__.py

@@ -1540,10 +1540,10 @@ def shutdown(self) -> None:
         for t in (self._prefetch_thread, self._sync_thread):
             if t and t.is_alive():
                 t.join(timeout=5.0)
-        # Flush any remaining messages
+        # Flush remaining messages and stop the manager's async writer.
         if self._manager and not (self._init_thread and self._init_thread.is_alive() and not self._session_initialized):
             try:
-                self._manager.flush_all()
+                self._manager.shutdown()
             except Exception:
                 pass
 

plugins/memory/honcho/session.py

@@ -7,6 +7,7 @@
 import re
 import logging
 import threading
+import time
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, TYPE_CHECKING
@@ -20,6 +21,7 @@
 
 # Sentinel to signal the async writer thread to shut down
 _ASYNC_SHUTDOWN = object()
+_CONTEXT_PREFETCH_DRAIN_TIMEOUT_S = 10.0
 _PEER_ID_HASH_LEN = 8
 _PEER_ID_HASH_ESCALATION_LENGTHS = (_PEER_ID_HASH_LEN, 12, 16, 24, 32, 64)
 
@@ -116,6 +118,9 @@ def __init__(
         # one source of truth; see __init__.py _do_session_init for the prewarm.
         self._context_cache: dict[str, dict] = {}
         self._prefetch_cache_lock = threading.Lock()
+        self._context_prefetch_threads: set[threading.Thread] = set()
+        self._context_prefetch_threads_lock = threading.Lock()
+        self._context_prefetch_shutting_down = False
         self._dialectic_reasoning_level: str = (
             config.dialectic_reasoning_level if config else "low"
         )
@@ -546,12 +551,37 @@ def flush_all(self) -> None:
                     break
 
     def shutdown(self) -> None:
-        """Gracefully shut down the async writer thread."""
+        """Gracefully shut down context prefetch and async writer threads."""
+        self._drain_context_prefetch_threads()
         if self._async_queue is not None and self._async_thread is not None:
             self.flush_all()
             self._async_queue.put(_ASYNC_SHUTDOWN)
             self._async_thread.join(timeout=10)
 
+    def _drain_context_prefetch_threads(self) -> None:
+        """Stop new context prefetches and wait boundedly for active HTTP calls."""
+        deadline = time.monotonic() + _CONTEXT_PREFETCH_DRAIN_TIMEOUT_S
+        with self._context_prefetch_threads_lock:
+            self._context_prefetch_shutting_down = True
+
+        while True:
+            with self._context_prefetch_threads_lock:
+                active = [t for t in self._context_prefetch_threads if t.is_alive()]
+            if not active:
+                return
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                logger.warning(
+                    "Honcho context prefetch shutdown timed out after %.1fs; "
+                    "%d request(s) remain active",
+                    _CONTEXT_PREFETCH_DRAIN_TIMEOUT_S,
+                    len(active),
+                )
+                return
+            for thread in active:
+                thread.join(timeout=max(0.0, deadline - time.monotonic()))
+
     def delete(self, key: str) -> bool:
         """Delete a session from local cache."""
         with self._cache_lock:
@@ -680,12 +710,31 @@ def prefetch_context(self, session_key: str, user_message: str | None = None) ->
         a synchronous HTTP round-trip blocking every response.
         """
         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()
+            try:
+                result = self.get_prefetch_context(session_key, user_message)
+                if result:
+                    self.set_context_result(session_key, result)
+            finally:
+                with self._context_prefetch_threads_lock:
+                    self._context_prefetch_threads.discard(threading.current_thread())
+
+        # Non-daemon is intentional. Provider shutdown joins these calls boundedly;
+        # if an HTTP request outlives that bound, Python must not finalize under
+        # a live native/socket stack.
+        thread = threading.Thread(
+            target=_run,
+            name="honcho-context-prefetch",
+            daemon=False,
+        )
+        with self._context_prefetch_threads_lock:
+            if self._context_prefetch_shutting_down:
+                return
+            self._context_prefetch_threads.add(thread)
+            try:
+                thread.start()
+            except Exception:
+                self._context_prefetch_threads.discard(thread)
+                raise
 
     def set_context_result(self, session_key: str, result: dict[str, str]) -> None:
         """Store a prefetched context result in a thread-safe way."""

tests/test_honcho_shutdown.py

@@ -0,0 +1,100 @@
+"""Regression tests for Honcho provider shutdown."""
+
+from __future__ import annotations
+
+import threading
+from types import SimpleNamespace
+
+from plugins.memory.honcho import HonchoMemoryProvider
+from plugins.memory.honcho.session import HonchoSessionManager
+
+
+def _async_config() -> SimpleNamespace:
+    return SimpleNamespace(
+        write_frequency="async",
+        dialectic_reasoning_level="low",
+        dialectic_dynamic=True,
+        dialectic_max_chars=600,
+        observation_mode="directional",
+        user_observe_me=True,
+        user_observe_others=True,
+        ai_observe_me=True,
+        ai_observe_others=True,
+        message_max_chars=25000,
+        dialectic_max_input_chars=10000,
+    )
+
+
+def test_provider_shutdown_stops_honcho_async_writer() -> None:
+    """Provider shutdown must not leave its session writer at interpreter exit."""
+    manager = HonchoSessionManager(config=_async_config())
+    provider = HonchoMemoryProvider()
+    provider._manager = manager
+
+    assert manager._async_thread is not None
+    assert manager._async_thread.is_alive()
+
+    try:
+        provider.shutdown()
+        assert not manager._async_thread.is_alive()
+    finally:
+        # Keep the regression test itself leak-free against the pre-fix code.
+        manager.shutdown()
+
+
+def test_provider_shutdown_waits_for_context_prefetch() -> None:
+    """CLI cleanup must not leave Honcho HTTP work at interpreter finalization."""
+    manager = HonchoSessionManager(config=_async_config())
+    provider = HonchoMemoryProvider()
+    provider._manager = manager
+    started = threading.Event()
+    release = threading.Event()
+    shutdown_done = threading.Event()
+
+    def slow_prefetch(
+        session_key: str, user_message: str | None = None
+    ) -> dict[str, str]:
+        started.set()
+        release.wait(timeout=2)
+        return {"representation": "ready"}
+
+    manager.get_prefetch_context = slow_prefetch  # type: ignore[method-assign]
+    manager.prefetch_context("session", "query")
+    assert started.wait(timeout=1)
+
+    def shut_down_provider() -> None:
+        try:
+            provider.shutdown()
+        finally:
+            shutdown_done.set()
+
+    shutdown_thread = threading.Thread(target=shut_down_provider)
+    shutdown_thread.start()
+    try:
+        assert not shutdown_done.wait(timeout=0.05)
+        release.set()
+        shutdown_thread.join(timeout=1)
+        assert shutdown_done.is_set()
+        assert not manager._context_prefetch_threads
+    finally:
+        release.set()
+        shutdown_thread.join(timeout=2)
+        manager.shutdown()
+
+
+def test_context_prefetch_is_rejected_after_shutdown() -> None:
+    manager = HonchoSessionManager(config=_async_config())
+    calls = 0
+
+    def record_prefetch(
+        session_key: str, user_message: str | None = None
+    ) -> dict[str, str]:
+        nonlocal calls
+        calls += 1
+        return {}
+
+    manager.get_prefetch_context = record_prefetch  # type: ignore[method-assign]
+    manager.shutdown()
+    manager.prefetch_context("session", "query")
+
+    assert calls == 0