From 53ddaff951c5f3713ed710aa1865bb55e4f1e8e7 Mon Sep 17 00:00:00 2001 From: c0d1ngHUB <268152820+c0d1ngHUB@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:09:37 +0200 Subject: [PATCH] fix(honcho): buffer turns during async session initialization --- plugins/memory/honcho/__init__.py | 113 ++++++++++++++++++---- tests/test_honcho_startup_fail_open.py | 126 +++++++++++++++++++++++++ 2 files changed, 221 insertions(+), 18 deletions(-) diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index 8ac56aec27dc..d8f068254126 100644 --- a/plugins/memory/honcho/__init__.py +++ b/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: diff --git a/tests/test_honcho_startup_fail_open.py b/tests/test_honcho_startup_fail_open.py index 2570a7a6c739..d86775b88d47 100644 --- a/tests/test_honcho_startup_fail_open.py +++ b/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()