fix(honcho): enforce saveMessages containment, honor writeFrequency, and clean up write-path shutdown

contributors/emails/carnie-bot@openclaw.local

@@ -0,0 +1,2 @@
+menhguin
+# agent bot from PR #82038

contributors/emails/daniel21436@hotmail.com

@@ -0,0 +1,2 @@
+strzhao
+# PR #81214 adoption

contributors/emails/dillontownsel@gmail.com

@@ -0,0 +1,2 @@
+dtownsel
+# PR #82130 adoption

plugins/memory/honcho/README.md

@@ -215,7 +215,7 @@ Pick **[e]** at the prompt to set the three keys directly instead of going throu
 | Key | Type | Default | Description |
 |-----|------|---------|-------------|
 | `writeFrequency` | string/int | `"async"` | `"async"` (background), `"turn"` (sync per turn), `"session"` (batch on end), or integer N (every N turns) |
-| `saveMessages` | bool | `true` | Persist messages to Honcho API |
+| `saveMessages` | bool | `true` | Persist messages to Honcho API. When `false`, all automatic writes are skipped — raw turns (`sync_turn`), conclusion mirroring (`on_memory_write`), and session-end/shutdown flushes — while read and tools paths stay fully functional. |
 
 ### Session Resolution
 

plugins/memory/honcho/__init__.py

@@ -29,6 +29,30 @@
 logger = logging.getLogger(__name__)
 
 
+# Gateway-internal notifications can arrive through the same user-role channel
+# as genuine user messages. They are execution metadata, not conversation, and
+# must never become durable personal memory. Keep this deliberately anchored:
+# a human discussing one of these strings mid-message is still valid input.
+_INTERNAL_GATEWAY_TURN_RE = re.compile(
+    r"^\s*(?:"
+    r"\[ASYNC (?:DELEGATION )?(?:BATCH )?COMPLETE[^\]]*\]|"
+    r"\[CONTEXT COMPACTION[^\]]*\]|"
+    r"\[CONTEXT SUMMARY\]:?|"
+    r"\[PRIOR CONTEXT[^\]]*\]|"
+    r"\[Your active task list was preserved across context compression\]|"
+    r"\[IMPORTANT: Background process \d+ matched watch pattern[^\n]*|"
+    r"A background fan-out of \d+ subagent\(s\) you dispatched earlier has finished\.|"
+    r"A background subagent you dispatched earlier has finished\."
+    r")",
+    re.IGNORECASE,
+)
+
+
+def _is_internal_gateway_turn(text: str) -> bool:
+    """Return True for machine-generated gateway/delegation notifications."""
+    return bool(_INTERNAL_GATEWAY_TURN_RE.match(text or ""))
+
+
 # ---------------------------------------------------------------------------
 # Tool schemas (moved from tools/honcho_tools.py)
 # ---------------------------------------------------------------------------
@@ -1390,9 +1414,20 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
 
         Messages exceeding the Honcho API limit (default 25k chars) are
         split into multiple messages with continuation markers.
+
+        Honors saveMessages: false — the provider then never persists raw
+        turns to Honcho (read/tools paths stay fully functional).
         """
         if self._cron_skipped:
             return
+        # ``saveMessages`` is the operator's hard write gate. Previously it
+        # was parsed into HonchoClientConfig but never enforced here, so a
+        # cached hybrid provider kept writing even after containment was set.
+        if self._config and not getattr(self._config, "save_messages", True):
+            return
+        if _is_internal_gateway_turn(user_content):
+            logger.debug("Honcho sync skipped machine-generated gateway turn")
+            return
         if self._recall_mode == "tools" and not self._session_ready():
             return
         if not self._session_ready():
@@ -1402,15 +1437,27 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         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()
+        # Skip only when the whole turn is empty. An interrupted or tool-only
+        # turn can legitimately have an empty assistant side; the user's
+        # message must still be persisted (the manager already drops
+        # empty-user turns upstream). Empty sides are skipped per-loop below
+        # so we never write empty-string messages either.
+        if not clean_user_content and not clean_assistant_content:
+            return
 
         def _sync():
             try:
                 session = self._manager.get_or_create(self._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)
+                if clean_user_content:
+                    for chunk in self._chunk_message(clean_user_content, msg_limit):
+                        session.add_message("user", chunk)
+                if clean_assistant_content:
+                    for chunk in self._chunk_message(clean_assistant_content, msg_limit):
+                        session.add_message("assistant", chunk)
+                # Route through save() so writeFrequency is honored —
+                # _flush_session() directly bypassed "session"/N batching
+                # and flushed every turn regardless of config.
+                self._manager.save(session)
             except Exception as e:
                 logger.debug("Honcho sync_turn failed: %s", e)
 
@@ -1439,6 +1486,11 @@ def on_memory_write(
             return
         if self._cron_skipped:
             return
+        # ``saveMessages`` is the operator's hard write gate; the memory-tool
+        # mirror is an automatic Honcho mutation path and must respect it too,
+        # otherwise containment would only cover conversation turns.
+        if self._config and not getattr(self._config, "save_messages", True):
+            return
         if self._recall_mode == "tools" and not self._session_ready():
             return
         if not self._session_ready():
@@ -1458,6 +1510,8 @@ 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 getattr(self._config, "save_messages", True):
+            return
         if not self._manager:
             return
         if not self._session_initialized and self._init_thread and self._init_thread.is_alive():
@@ -1610,10 +1664,26 @@ 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
-        if self._manager and not (self._init_thread and self._init_thread.is_alive() and not self._session_initialized):
+        manager = self._manager
+        if manager and self._init_thread and self._init_thread.is_alive() and not self._session_initialized:
+            manager = None
+        # Honors saveMessages: false — skip persistence, but thread cleanup
+        # still runs: the session manager's async-writer thread must be
+        # joined either way so daemon threads aren't left blocked in httpx
+        # I/O during interpreter finalization.
+        if not getattr(self._config, "save_messages", True):
+            if manager:
+                try:
+                    manager.stop_async_writer()
+                except Exception:
+                    pass
+            return
+        if manager:
             try:
-                self._manager.flush_all()
+                # manager.shutdown() = flush_all() + join the async-writer
+                # thread. Previously only flush_all() ran here, leaving the
+                # writer thread alive at exit.
+                manager.shutdown()
             except Exception:
                 pass
 

plugins/memory/honcho/session.py

@@ -538,6 +538,19 @@ def _generated_runtime_peer_id(self, prefix: str, runtime_id: str) -> str:
             return f"{sanitized_peer_id}-{digest}"
         return sanitized_peer_id
 
+    def _declared_owner_peer_id(self) -> str | None:
+        """Peer ID of the install owner, or None when no owner is declared.
+
+        The owner is the identity setup writes as ``peerName``. A runtime
+        gateway identity is the owner only when an alias maps it onto that
+        peer — which _resolve_user_peer_id already does, so callers can
+        compare a session's resolved user peer against this value.
+        """
+        peer_name = getattr(self._config, "peer_name", None) if self._config else None
+        if peer_name and str(peer_name).strip():
+            return self._sanitize_id(str(peer_name).strip())
+        return None
+
     def _resolve_user_peer_id(self, key: str) -> str:
         """Resolve the Honcho user peer ID for this manager/session."""
         pin_peer_name = (
@@ -771,6 +784,18 @@ def _ensure_async_writer(self) -> None:
                 )
                 self._async_thread.start()
 
+    def stop_async_writer(self) -> None:
+        """Stop the async writer thread WITHOUT flushing pending messages.
+
+        Used on shutdown when persistence is disabled (saveMessages: false):
+        the thread must still be joined so process exit is clean, but nothing
+        may be written.
+        """
+        if self._async_queue is not None:
+            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)
+
     def shutdown(self) -> None:
         """Gracefully shut down the async writer thread."""
         if self._async_queue is not None:
@@ -1103,6 +1128,38 @@ def migrate_memory_files(self, session_key: str, memory_dir: str) -> bool:
             logger.warning("No Honcho session cached for '%s', skipping memory migration", session_key)
             return False
 
+        # Only migrate the owner-describing memory files (MEMORY.md / USER.md)
+        # when the session's user peer IS the install owner. Otherwise a
+        # non-owner triggering a new session (e.g. any other human in a shared
+        # Slack/Discord channel) gets the owner's full profile files uploaded
+        # under the NON-OWNER's peer, and Honcho's deriver attributes the
+        # owner's facts to that person. SOUL.md describes the agent, not a
+        # human, but skipping it here too keeps the migration owner-scoped.
+        #
+        # The owner is a CONFIG fact — the declared peerName — never a
+        # re-resolution of the session's own peer: _resolve_user_peer_id
+        # answers "who is this session's user", so comparing its output to
+        # session.user_peer_id compares the triggering user to themselves
+        # and passes for the non-owner too.
+        owner_peer_id = self._declared_owner_peer_id()
+        if owner_peer_id is not None:
+            session_is_owner = session.user_peer_id == owner_peer_id
+        else:
+            # No declared owner. Without a runtime identity this is the
+            # single-operator path (peer id from config defaults or the
+            # session key) and the files describe that operator. With a
+            # runtime identity the session belongs to whoever messaged
+            # through the gateway — nobody can be proven to be the owner.
+            session_is_owner = not self._runtime_user_ids()
+        if not session_is_owner:
+            logger.info(
+                "Skipping memory-file migration: session user peer '%s' is not the "
+                "declared owner (peerName=%s)",
+                session.user_peer_id,
+                owner_peer_id or "unset",
+            )
+            return False
+
         uploaded = False
         files = [
             (