fix(honcho): buffer turns during async session initialization

plugins/memory/honcho/__init__.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+from collections import deque
 import json
 import logging
 import re
@@ -237,6 +238,9 @@
 class HonchoMemoryProvider(MemoryProvider):
     """Honcho AI-native memory with dialectic Q&A and persistent user modeling."""
 
+    _PENDING_TURN_MAX = 32
+    _PENDING_TURN_CHARS_MAX = 1_000_000
+
     def backup_paths(self) -> List[str]:
         """Honcho keeps its peer/session config under ~/.honcho when no
         profile-local honcho.json exists (see client.resolve_config_path)."""
@@ -292,6 +296,11 @@ def __init__(self, query_rewriter: Optional[Callable[[str], str]] = None):
         self._init_lock = threading.Lock()
         self._init_error = ""
 
+        self._pending_turns = deque()
+        self._pending_turn_chars = 0
+        self._pending_turns_lock = threading.Lock()
+        self._pending_turns_closing = False
+
         # Cron and flush contexts disable the plugin entirely.
         self._cron_skipped = False
 
@@ -450,10 +459,12 @@ def _run() -> None:
                     self._lazy_init_kwargs = None
                     self._lazy_init_session_id = None
                     self._init_error = ""
+                    self._flush_pending_turns()
                 except Exception as e:
                     self._init_error = str(e)
                     self._manager = None
                     logger.warning("Honcho background session init failed: %s", e)
+                    self._drop_pending_turns("initialization failed")
 
             self._init_thread = threading.Thread(
                 target=_run,
@@ -1329,37 +1340,95 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         """Record the conversation turn in Honcho (non-blocking).
 
         Messages exceeding the Honcho API limit (default 25k chars) are
-        split into multiple messages with continuation markers.
+        split into multiple messages with continuation markers. While hybrid
+        startup is still creating the remote session, completed turns are kept
+        in a bounded FIFO and flushed by one writer after initialization.
         """
         if self._cron_skipped:
             return
         if self._recall_mode == "tools" and not self._session_ready():
             return
+        clean_user_content = sanitize_context(user_content or "").strip()
+        clean_assistant_content = sanitize_context(assistant_content or "").strip()
+        if not self._buffer_pending_turn(clean_user_content, clean_assistant_content):
+            return
         if not self._session_ready():
             self._start_session_init_background()
             return
+        self._flush_pending_turns()
+
+    def _buffer_pending_turn(self, user_content: str, assistant_content: str) -> bool:
+        """Append one completed turn to the bounded FIFO without logging content."""
+        turn_chars = len(user_content) + len(assistant_content)
+        with self._pending_turns_lock:
+            if self._pending_turns_closing:
+                logger.warning("Honcho pending turn dropped: provider is shutting down")
+                return False
+            if (
+                len(self._pending_turns) >= self._PENDING_TURN_MAX
+                or self._pending_turn_chars + turn_chars > self._PENDING_TURN_CHARS_MAX
+            ):
+                logger.warning(
+                    "Honcho pending turn dropped: buffer full (turns=%d chars=%d)",
+                    len(self._pending_turns), self._pending_turn_chars,
+                )
+                return False
+            self._pending_turns.append((user_content, assistant_content, turn_chars))
+            self._pending_turn_chars += turn_chars
+            logger.debug(
+                "Honcho pending turn buffered (turns=%d chars=%d)",
+                len(self._pending_turns), self._pending_turn_chars,
+            )
+            return True
 
-        msg_limit = self._config.message_max_chars if self._config else 25000
-        clean_user_content = sanitize_context(user_content or "").strip()
-        clean_assistant_content = sanitize_context(assistant_content or "").strip()
+    def _flush_pending_turns(self) -> None:
+        """Start the sole FIFO writer after a ready session becomes available."""
+        if not self._session_ready():
+            return
+        with self._pending_turns_lock:
+            if not self._pending_turns:
+                return
+            if self._sync_thread and self._sync_thread.is_alive():
+                return
+            self._sync_thread = threading.Thread(
+                target=self._drain_pending_turns,
+                daemon=True,
+                name="honcho-sync",
+            )
+            self._sync_thread.start()
 
-        def _sync():
+    def _drain_pending_turns(self) -> None:
+        """Flush buffered turns in FIFO order; only this thread writes turns."""
+        msg_limit = self._config.message_max_chars if self._config else 25000
+        while True:
+            with self._pending_turns_lock:
+                if not self._pending_turns:
+                    self._sync_thread = None
+                    return
+                user_content, assistant_content, turn_chars = self._pending_turns.popleft()
+                self._pending_turn_chars -= turn_chars
             try:
                 session = self._manager.get_or_create(self._session_key)
-                for chunk in self._chunk_message(clean_user_content, msg_limit):
+                for chunk in self._chunk_message(user_content, msg_limit):
                     session.add_message("user", chunk)
-                for chunk in self._chunk_message(clean_assistant_content, msg_limit):
+                for chunk in self._chunk_message(assistant_content, msg_limit):
                     session.add_message("assistant", chunk)
                 self._manager._flush_session(session)
             except Exception as e:
-                logger.debug("Honcho sync_turn failed: %s", e)
-
-        if self._sync_thread and self._sync_thread.is_alive():
-            self._sync_thread.join(timeout=5.0)
-        self._sync_thread = threading.Thread(
-            target=_sync, daemon=True, name="honcho-sync"
-        )
-        self._sync_thread.start()
+                logger.debug("Honcho pending turn flush failed: %s", e)
+
+    def _drop_pending_turns(self, reason: str) -> None:
+        """Discard buffered turns only on terminal teardown/init failure."""
+        with self._pending_turns_lock:
+            count = len(self._pending_turns)
+            chars = self._pending_turn_chars
+            self._pending_turns.clear()
+            self._pending_turn_chars = 0
+        if count:
+            logger.warning(
+                "Honcho pending turns dropped: %s (turns=%d chars=%d)",
+                reason, count, chars,
+            )
 
     def on_memory_write(
         self,
@@ -1398,13 +1467,14 @@ def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
         """Flush all pending messages to Honcho on session end."""
         if self._cron_skipped:
             return
-        if not self._manager:
-            return
         if not self._session_initialized and self._init_thread and self._init_thread.is_alive():
-            return
+            self._init_thread.join(timeout=5.0)
+        self._flush_pending_turns()
         # Wait for pending sync
         if self._sync_thread and self._sync_thread.is_alive():
             self._sync_thread.join(timeout=10.0)
+        if not self._manager:
+            return
         try:
             self._manager.flush_all()
         except Exception as e:
@@ -1537,9 +1607,16 @@ 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:
+        with self._pending_turns_lock:
+            self._pending_turns_closing = True
+        if self._init_thread and self._init_thread.is_alive():
+            self._init_thread.join(timeout=5.0)
+        self._flush_pending_turns()
         for t in (self._prefetch_thread, self._sync_thread):
             if t and t.is_alive():
                 t.join(timeout=5.0)
+        if not self._session_ready():
+            self._drop_pending_turns("shutdown before initialization completed")
         # Flush any remaining messages
         if self._manager and not (self._init_thread and self._init_thread.is_alive() and not self._session_initialized):
             try:

tests/test_honcho_startup_fail_open.py

@@ -45,6 +45,97 @@ def _configured_tools_config(*, init_on_session_start: bool = False) -> _FakeHon
     return cfg
 
 
+
+class _RecordingSession:
+    def __init__(self):
+        self.messages = []
+
+    def add_message(self, role, content):
+        self.messages.append((role, content))
+
+
+class _RecordingManager:
+    def __init__(self):
+        self.session = _RecordingSession()
+        self.flushes = 0
+
+    def get_or_create(self, session_key):
+        return self.session
+
+    def _flush_session(self, session):
+        self.flushes += 1
+
+    def flush_all(self):
+        self.flushes += 1
+
+
+def _wait_for_sync(provider):
+    sync_thread = provider._sync_thread
+    if sync_thread:
+        sync_thread.join(timeout=1)
+
+
+def test_honcho_buffers_turn_until_background_init_succeeds(monkeypatch):
+    provider = HonchoMemoryProvider()
+    cfg = _configured_hybrid_config()
+    manager = _RecordingManager()
+    release = threading.Event()
+
+    monkeypatch.setattr(
+        "plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
+        lambda: cfg,
+    )
+
+    def delayed_session_init(self, cfg, session_id, **kwargs):
+        release.wait(timeout=5)
+        self._manager = manager
+        self._session_key = "test-session"
+        self._session_initialized = True
+
+    monkeypatch.setattr(HonchoMemoryProvider, "_do_session_init", delayed_session_init)
+    provider.initialize("session-1", platform="api_server")
+    provider.sync_turn("buffered user", "buffered assistant")
+
+    assert len(provider._pending_turns) == 1
+    release.set()
+    provider._init_thread.join(timeout=1)
+    _wait_for_sync(provider)
+
+    assert manager.session.messages == [
+        ("user", "buffered user"),
+        ("assistant", "buffered assistant"),
+    ]
+
+
+def test_honcho_flushes_buffered_turns_fifo_after_init():
+    provider = HonchoMemoryProvider()
+    provider._config = _configured_hybrid_config()
+    provider._manager = _RecordingManager()
+    provider._session_key = "test-session"
+
+    provider.sync_turn("one", "one reply")
+    provider.sync_turn("two", "two reply")
+    provider._session_initialized = True
+    provider._flush_pending_turns()
+    _wait_for_sync(provider)
+
+    assert provider._manager.session.messages == [
+        ("user", "one"), ("assistant", "one reply"),
+        ("user", "two"), ("assistant", "two reply"),
+    ]
+
+
+def test_honcho_pending_turn_buffer_drops_new_turn_when_full(caplog):
+    provider = HonchoMemoryProvider()
+    provider._config = _configured_hybrid_config()
+    for index in range(provider._PENDING_TURN_MAX):
+        assert provider._buffer_pending_turn(f"user-{index}", "assistant")
+
+    assert not provider._buffer_pending_turn("overflow", "assistant")
+    assert len(provider._pending_turns) == provider._PENDING_TURN_MAX
+    assert "buffer full" in caplog.text
+    assert "overflow" not in caplog.text
+
 def test_honcho_hybrid_initialize_returns_without_waiting_for_session_init(monkeypatch):
     """Slow Honcho session creation must not block agent startup."""
     provider = HonchoMemoryProvider()
@@ -313,6 +404,41 @@ def _flush_session(self, session):
     assert provider._session_initialized is True
 
 
+
+
+def test_honcho_shutdown_flushes_pending_turn_once_after_init():
+    provider = HonchoMemoryProvider()
+    provider._config = _configured_hybrid_config()
+    provider._manager = _RecordingManager()
+    provider._session_key = "test-session"
+    provider.sync_turn("user", "assistant")
+    provider._session_initialized = True
+
+    provider.shutdown()
+    provider.shutdown()
+
+    assert provider._manager.session.messages == [("user", "user"), ("assistant", "assistant")]
+
+
+def test_honcho_failed_background_init_drops_pending_turns(monkeypatch, caplog):
+    provider = HonchoMemoryProvider()
+    cfg = _configured_hybrid_config()
+
+    monkeypatch.setattr(
+        "plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
+        lambda: cfg,
+    )
+    monkeypatch.setattr(
+        HonchoMemoryProvider, "_do_session_init",
+        lambda self, cfg, session_id, **kwargs: (_ for _ in ()).throw(RuntimeError("unavailable")),
+    )
+    provider.initialize("session-1", platform="api_server")
+    provider.sync_turn("user", "assistant")
+    provider._init_thread.join(timeout=1)
+
+    assert not provider._pending_turns
+    assert "initialization failed" in caplog.text
+
 def test_honcho_system_prompt_advertises_active_while_background_init_runs(monkeypatch):
     """Prompt metadata should not require a completed network session."""
     provider = HonchoMemoryProvider()