feat(honcho): adaptive HTTP timeout + circuit breaker + sync worker (draft)

agent/memory_manager.py

@@ -63,6 +63,117 @@ def sanitize_context(text: str) -> str:
     return text
 
 
+class StreamingContextScrubber:
+    """Stateful scrubber for streaming text that may contain split memory-context spans.
+
+    The one-shot ``sanitize_context`` regex cannot survive chunk boundaries:
+    a ``<memory-context>`` opened in one delta and closed in a later delta
+    leaks its payload to the UI because the non-greedy block regex needs
+    both tags in one string.  This scrubber runs a small state machine
+    across deltas, holding back partial-tag tails and discarding
+    everything inside a span (including the system-note line).
+
+    Usage::
+
+        scrubber = StreamingContextScrubber()
+        for delta in stream:
+            visible = scrubber.feed(delta)
+            if visible:
+                emit(visible)
+        trailing = scrubber.flush()  # at end of stream
+        if trailing:
+            emit(trailing)
+
+    The scrubber is re-entrant per agent instance.  Callers building new
+    top-level responses (new turn) should create a fresh scrubber or call
+    ``reset()``.
+    """
+
+    _OPEN_TAG = "<memory-context>"
+    _CLOSE_TAG = "</memory-context>"
+
+    def __init__(self) -> None:
+        self._in_span: bool = False
+        self._buf: str = ""
+
+    def reset(self) -> None:
+        self._in_span = False
+        self._buf = ""
+
+    def feed(self, text: str) -> str:
+        """Return the visible portion of ``text`` after scrubbing.
+
+        Any trailing fragment that could be the start of an open/close tag
+        is held back in the internal buffer and surfaced on the next
+        ``feed()`` call or discarded/emitted by ``flush()``.
+        """
+        if not text:
+            return ""
+        buf = self._buf + text
+        self._buf = ""
+        out: list[str] = []
+
+        while buf:
+            if self._in_span:
+                idx = buf.lower().find(self._CLOSE_TAG)
+                if idx == -1:
+                    # Hold back a potential partial close tag; drop the rest
+                    held = self._max_partial_suffix(buf, self._CLOSE_TAG)
+                    self._buf = buf[-held:] if held else ""
+                    return "".join(out)
+                # Found close — skip span content + tag, continue
+                buf = buf[idx + len(self._CLOSE_TAG):]
+                self._in_span = False
+            else:
+                idx = buf.lower().find(self._OPEN_TAG)
+                if idx == -1:
+                    # No open tag — hold back a potential partial open tag
+                    held = self._max_partial_suffix(buf, self._OPEN_TAG)
+                    if held:
+                        out.append(buf[:-held])
+                        self._buf = buf[-held:]
+                    else:
+                        out.append(buf)
+                    return "".join(out)
+                # Emit text before the tag, enter span
+                if idx > 0:
+                    out.append(buf[:idx])
+                buf = buf[idx + len(self._OPEN_TAG):]
+                self._in_span = True
+
+        return "".join(out)
+
+    def flush(self) -> str:
+        """Emit any held-back buffer at end-of-stream.
+
+        If we're still inside an unterminated span the remaining content is
+        discarded (safer: leaking partial memory context is worse than a
+        truncated answer).  Otherwise the held-back partial-tag tail is
+        emitted verbatim (it turned out not to be a real tag).
+        """
+        if self._in_span:
+            self._buf = ""
+            self._in_span = False
+            return ""
+        tail = self._buf
+        self._buf = ""
+        return tail
+
+    @staticmethod
+    def _max_partial_suffix(buf: str, tag: str) -> int:
+        """Return the length of the longest buf-suffix that is a tag-prefix.
+
+        Case-insensitive.  Returns 0 if no suffix could start the tag.
+        """
+        tag_lower = tag.lower()
+        buf_lower = buf.lower()
+        max_check = min(len(buf_lower), len(tag_lower) - 1)
+        for i in range(max_check, 0, -1):
+            if tag_lower.startswith(buf_lower[-i:]):
+                return i
+        return 0
+
+
 def build_memory_context_block(raw_context: str) -> str:
     """Wrap prefetched memory in a fenced block with system note.
 

gateway/run.py

@@ -8253,6 +8253,7 @@ async def _enrich_message_with_vision(
             The enriched message string with vision descriptions prepended.
         """
         from tools.vision_tools import vision_analyze_tool
+        from agent.memory_manager import sanitize_context
 
         analysis_prompt = (
             "Describe everything visible in this image in thorough detail. "
@@ -8271,6 +8272,14 @@ async def _enrich_message_with_vision(
                 result = json.loads(result_json)
                 if result.get("success"):
                     description = result.get("analysis", "")
+                    # The auxiliary vision LLM can echo injected system-prompt
+                    # memory context back into its output (#5719).  Scrub any
+                    # <memory-context> fences and the "## Honcho Context"
+                    # section before the description lands in a user-visible
+                    # message.
+                    description = sanitize_context(description)
+                    if "## Honcho Context" in description:
+                        description = description.split("## Honcho Context", 1)[0].rstrip()
                     enriched_parts.append(
                         f"[The user sent an image~ Here's what I can see:\n{description}]\n"
                         f"[If you need a closer look, use vision_analyze with "

hermes_state.py

@@ -22,6 +22,8 @@
 import threading
 import time
 from pathlib import Path
+
+from agent.memory_manager import sanitize_context
 from hermes_constants import get_hermes_home
 from typing import Any, Callable, Dict, List, Optional, TypeVar
 
@@ -1119,7 +1121,10 @@ def get_messages_as_conversation(self, session_id: str) -> List[Dict[str, Any]]:
             rows = cursor.fetchall()
         messages = []
         for row in rows:
-            msg = {"role": row["role"], "content": row["content"]}
+            content = row["content"]
+            if row["role"] in {"user", "assistant"} and isinstance(content, str):
+                content = sanitize_context(content).strip()
+            msg = {"role": row["role"], "content": content}
             if row["tool_call_id"]:
                 msg["tool_call_id"] = row["tool_call_id"]
             if row["tool_name"]:

plugins/memory/honcho/__init__.py

@@ -22,7 +22,14 @@
 import time
 from typing import Any, Dict, List, Optional
 
+from agent.memory_manager import sanitize_context
 from agent.memory_provider import MemoryProvider
+from plugins.memory.honcho.sync_worker import (
+    CircuitBreaker,
+    HonchoLatencyTracker,
+    SyncTask,
+    SyncWorker,
+)
 from tools.registry import tool_error
 
 logger = logging.getLogger(__name__)
@@ -194,7 +201,22 @@ def __init__(self):
         self._prefetch_result = ""
         self._prefetch_lock = threading.Lock()
         self._prefetch_thread: Optional[threading.Thread] = None
-        self._sync_thread: Optional[threading.Thread] = None
+
+        # Post-response write path (sync_turn / on_memory_write).  See
+        # plugins/memory/honcho/sync_worker.py.  The tracker + breaker are
+        # shared with the Honcho SDK client so adaptive timeouts and
+        # degraded-mode behaviour are consistent across the plugin.
+        self._latency_tracker = HonchoLatencyTracker()
+        self._breaker = CircuitBreaker()
+        self._sync_worker = SyncWorker(
+            latency_tracker=self._latency_tracker,
+            breaker=self._breaker,
+            thread_name="honcho-sync-worker",
+        )
+        # Durable backlog of tasks that couldn't reach Honcho (breaker open
+        # or queue overflow).  Drained on recovery — see _drain_backlog().
+        self._backlog: List[SyncTask] = []
+        self._backlog_lock = threading.Lock()
 
         # B1: recall_mode — set during initialize from config
         self._recall_mode = "hybrid"  # "context", "tools", or "hybrid"
@@ -1056,8 +1078,82 @@ def _chunk_message(content: str, limit: int) -> list[str]:
 
         return chunks
 
+    # -- backlog management (Layer 3) ----------------------------------------
+
+    _BACKLOG_MAX = 256
+
+    def _enqueue_with_backlog(self, task: SyncTask) -> None:
+        """Submit a task to the worker with backlog fall-through on defer.
+
+        Wraps the caller's task with a failure hook that captures the
+        task itself (not just the error) so it can be appended to the
+        durable backlog when the breaker is open or the queue is full.
+        """
+        original_on_failure = task.on_failure
+
+        def _on_failure(error: BaseException) -> None:
+            reason = str(error)
+            # Only backlog tasks that were deferred, not ones that crashed
+            # inside Honcho itself — those are unlikely to succeed on replay.
+            if any(marker in reason for marker in (
+                "circuit breaker open",
+                "sync queue full",
+                "sync queue overflow",
+                "shutting down",
+            )):
+                with self._backlog_lock:
+                    if len(self._backlog) < self._BACKLOG_MAX:
+                        self._backlog.append(task)
+                    else:
+                        logger.debug("Honcho backlog full; dropping %s", task.name)
+            else:
+                logger.debug(
+                    "Honcho sync task %s failed (not backlogged): %s",
+                    task.name, error,
+                )
+            if original_on_failure is not None:
+                try:
+                    original_on_failure(error)
+                except Exception:
+                    pass
+
+        task.on_failure = _on_failure
+        self._sync_worker.enqueue(task)
+
+    def _drain_backlog_if_healthy(self) -> None:
+        """Opportunistic replay of backlogged tasks when the breaker closes.
+
+        Called from the happy path of ``sync_turn``; never blocks the
+        user.  Walks the backlog, re-enqueueing everything on the worker
+        — if the breaker is still open, tasks will bounce back into
+        the backlog via their on_failure handler.
+
+        Also nudges the Honcho client's HTTP timeout toward the tracker's
+        observed p95 so stalled backends fail fast on subsequent calls
+        instead of burning 30s per request.
+        """
+        if self._breaker.state != self._breaker.STATE_CLOSED:
+            return
+        # Layer 2: adaptive timeout rebuild (cheap no-op if below threshold).
+        try:
+            from plugins.memory.honcho.client import rebuild_honcho_client_with_timeout
+            rebuild_honcho_client_with_timeout(self._latency_tracker.timeout())
+        except Exception as e:
+            logger.debug("Honcho timeout rebuild skipped: %s", e)
+        with self._backlog_lock:
+            if not self._backlog:
+                return
+            pending = self._backlog
+            self._backlog = []
+        for task in pending:
+            self._sync_worker.enqueue(task)
+
     def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
-        """Record the conversation turn in Honcho (non-blocking).
+        """Record the conversation turn in Honcho (fire-and-forget).
+
+        Enqueues the sync on the persistent worker thread and returns
+        immediately.  Callers never wait on Honcho — the run_conversation
+        return path is fully decoupled from post-response writes.
 
         Messages exceeding the Honcho API limit (default 25k chars) are
         split into multiple messages with continuation markers.
@@ -1068,52 +1164,63 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
             return
 
         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()
+        session_key = self._session_key
 
         def _sync():
-            try:
-                session = self._manager.get_or_create(self._session_key)
-                for chunk in self._chunk_message(user_content, msg_limit):
-                    session.add_message("user", chunk)
-                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"
+            session = self._manager.get_or_create(session_key)
+            for chunk in self._chunk_message(clean_user_content, msg_limit):
+                session.add_message("user", chunk)
+            for chunk in self._chunk_message(clean_assistant_content, msg_limit):
+                session.add_message("assistant", chunk)
+            self._manager._flush_session(session)
+
+        task = SyncTask(
+            fn=_sync,
+            name="sync_turn",
         )
-        self._sync_thread.start()
+        self._enqueue_with_backlog(task)
+        # If the breaker transitioned back to closed between turns, try to
+        # drain anything that piled up while Honcho was unreachable.
+        self._drain_backlog_if_healthy()
 
     def on_memory_write(self, action: str, target: str, content: str) -> None:
-        """Mirror built-in user profile writes as Honcho conclusions."""
+        """Mirror built-in user profile writes as Honcho conclusions.
+
+        Enqueued on the shared sync worker so every post-response write
+        path (turn sync + conclusion mirror) observes the same breaker
+        and backlog.
+        """
         if action != "add" or target != "user" or not content:
             return
         if self._cron_skipped:
             return
         if not self._manager or not self._session_key:
             return
 
+        session_key = self._session_key
+        payload = content
+
         def _write():
-            try:
-                self._manager.create_conclusion(self._session_key, content)
-            except Exception as e:
-                logger.debug("Honcho memory mirror failed: %s", e)
+            self._manager.create_conclusion(session_key, payload)
 
-        t = threading.Thread(target=_write, daemon=True, name="honcho-memwrite")
-        t.start()
+        task = SyncTask(
+            fn=_write,
+            name="memory_mirror",
+        )
+        self._enqueue_with_backlog(task)
 
     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
-        # Wait for pending sync
-        if self._sync_thread and self._sync_thread.is_alive():
-            self._sync_thread.join(timeout=10.0)
+        # Wait briefly for any in-flight sync tasks to drain.  We can't
+        # block session-end indefinitely, but giving the worker 10s to
+        # finish a pending turn-sync matches the previous behaviour.
+        self._sync_worker.shutdown(timeout=10.0)
         try:
             self._manager.flush_all()
         except Exception as e:
@@ -1233,9 +1340,10 @@ 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:
-        for t in (self._prefetch_thread, self._sync_thread):
-            if t and t.is_alive():
-                t.join(timeout=5.0)
+        # Drain the prefetch thread (legacy, unchanged) + the sync worker.
+        if self._prefetch_thread and self._prefetch_thread.is_alive():
+            self._prefetch_thread.join(timeout=5.0)
+        self._sync_worker.shutdown(timeout=5.0)
         # Flush any remaining messages
         if self._manager:
             try: