fix(honcho): every bot author gets its own peer or its turn is skipped

A gateway platform marks a bot sender with its raw user id and a bot flag, never a `bot:` id.
`resolve_author_peer_id` treated that author as a human, so `pinUserPeer` collapsed a bot onto
`peerName` and an unresolved peer opened the a2a session under the human's peer. The resolver now
takes `is_bot` and gives every bot its own peer. `sync_turn` skips the turn when no peer resolves
or when the peer equals this agent's `aiPeer`.

The a2a session key carries an eight-character digest of the author id, so two ids that sanitize
alike stay in separate sessions. During a bot-authored turn `honcho_conclude` and `honcho_profile`
refuse writes and the built-in memory mirror is skipped, because conclusions and cards describe
the human. The README paragraph on bot DMs now matches the code.

plugins/memory/honcho/README.md

@@ -276,7 +276,7 @@ If `sessionPeerPrefix` is `true`, the peer name is prepended: `alice-hermes-agen
 
 #### Bot DMs (`a2aSessions`)
 
-In bot mode another Hermes profile can DM this agent. The relay marks that turn with author `bot:<profile>`, and the turn never reaches the human's session. With `a2aSessions: true` (default) the turn is written into `<session>:a2a:<profile>`: the sender's message under the sender's peer, the reply under this agent's `aiPeer`. The sender's peer is the `userPeerAliases` entry for `bot:<profile>` if one exists, else the profile name, which is the sender's own default `aiPeer` in a shared workspace. `runtimePeerPrefix` is not applied to bot ids. `pinUserPeer: true` collapses bot authors onto `peerName` like every other author. With `a2aSessions: false` bot-authored turns are skipped. Recall still reads the human's session only.
+In bot mode another Hermes profile can DM this agent. The relay marks that turn with author `bot:<profile>`. A gateway platform marks a bot sender with its platform user id and a bot flag. Either way the turn never reaches the human's session. With `a2aSessions: true` (default) the turn is written into `<session>:a2a:<sanitized id>-<8-char digest>`: the sender's message under the sender's peer, the reply under this agent's `aiPeer`. A `bot:<profile>` sender's peer is the `userPeerAliases` entry for `bot:<profile>` if one exists, else the profile name with no `runtimePeerPrefix`, which is the sender's own default `aiPeer` in a shared workspace. A platform bot resolves like any other runtime user: alias, then prefix. `pinUserPeer` never collapses a bot onto `peerName`. A bot whose peer would equal this agent's `aiPeer` is skipped, and so is every bot turn when `a2aSessions: false`. During a bot-authored turn `honcho_conclude` and `honcho_profile` refuse writes, because conclusions and cards describe the human. Recall still reads the human's session only.
 
 #### What each strategy produces
 

plugins/memory/honcho/__init__.py

@@ -8,6 +8,7 @@ Config chain: $HERMES_HOME/honcho.json -> ~/.honcho/config.json -> env vars.
 from __future__ import annotations
 
 import contextlib
+import hashlib
 import json
 import logging
 import re
@@ -555,9 +556,10 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
     _is_trivial_prompt = staticmethod(is_trivial_prompt)
 
     def identity_signature(self) -> Dict[str, Any]:
-        """Identity-mapping values from honcho.json that must bust a cached gateway agent when they
-        change. Reads config only, never the network, and memoizes on the file's mtime and size so the
-        per-message call stays a stat. ``{}`` when the config cannot be read."""
+        """Identity-mapping values from honcho.json that bust a cached gateway agent when they change.
+
+        Memoized on the file's mtime and size, so the per-message call is one stat. ``{}`` when the
+        config cannot be read."""
         try:
             path = resolve_config_path()
             try:
@@ -638,7 +640,7 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
     ) -> None:
         """Record the conversation turn in Honcho (non-blocking), chunking messages that
         exceed the Honcho API limit. Honors saveMessages: false. ``turn_author`` names who wrote
-        the user side; the ``on_turn_start`` stash is the fallback for callers that never pass it.
+        the user side. The ``on_turn_start`` stash is the fallback for callers that never pass it.
         A bot author's turn is written into that bot's own a2a session, never the human's."""
         if not self._writes_enabled():
             return
@@ -667,11 +669,14 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
             if not author_id:
                 logger.debug("Honcho sync skipped a bot-authored turn that named no author id")
                 return
-            bot_peer_id = self._manager.resolve_author_peer_id(self._session_key, author_id, author.get("name"))
+            bot_peer_id = self._manager.resolve_author_peer_id(
+                self._session_key, author_id, author.get("name"), is_bot=True)
+            if not bot_peer_id or bot_peer_id == self._manager.assistant_peer_id():
+                logger.debug("Honcho sync skipped a bot-authored turn: author %s has no peer of its own", author_id)
+                return
             session_key = self._a2a_session_key(author_id)
             # The bot is the a2a session's own user peer, so its messages need no per-message author.
-            if bot_peer_id:
-                session_kwargs["user_peer_id"] = bot_peer_id
+            session_kwargs["user_peer_id"] = bot_peer_id
             author_peer_id = None
         else:
             session_key = self._session_key
@@ -692,11 +697,19 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
         self._sync_thread = self._spawn_write(_sync, "honcho-sync", "Honcho sync_turn failed: %s")
 
     def _a2a_session_key(self, author_id: str) -> str:
-        """Honcho session for one sender bot's DMs into this agent, stable across turns. Rooms already
-        get their own session by title, so only DMs reroute."""
-        key = f"{self._session_key}:a2a:{re.sub(r'[^a-zA-Z0-9_-]', '-', author_id)}"
+        """Honcho session for one sender bot's turns into this agent, stable across turns.
+
+        The digest keeps two ids apart when sanitizing would make them equal."""
+        digest = hashlib.sha256(author_id.encode("utf-8")).hexdigest()[:8]
+        key = f"{self._session_key}:a2a:{re.sub(r'[^a-zA-Z0-9_-]', '-', author_id)}-{digest}"
         return HonchoClientConfig._enforce_session_id_limit(key, key)
 
+    def _bot_turn_write_refusal(self) -> Optional[str]:
+        """Refusal for memory writes while a bot-authored turn runs. Conclusions and cards describe the human."""
+        if self._turn_author.get("is_bot"):
+            return tool_error("Honcho memory writes are off during a bot-to-bot turn. Conclusions and profile edits describe the human.")
+        return None
+
     @staticmethod
     def _spawn_write(fn: Callable[[], None], name: str, fail_msg: str) -> threading.Thread:
         """Run a Honcho write off-thread; failures are debug-logged, never raised into the turn."""
@@ -717,6 +730,9 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
         for interface compatibility, not yet threaded into the conclusion payload)."""
         if action != "add" or target != "user" or not content:
             return
+        if self._turn_author.get("is_bot"):
+            logger.debug("Honcho memory mirror skipped during a bot-authored turn")
+            return
         if not self._writes_enabled() or not self._ready_or_kick_init():
             return
         self._memwrite_thread = self._spawn_write(lambda: self._manager.create_conclusion(self._session_key, content),
@@ -769,6 +785,8 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
     def _tool_profile(self, args: dict) -> str:
         peer = args.get("peer", "user")
         if card_update := args.get("card"):
+            if refusal := self._bot_turn_write_refusal():
+                return refusal
             result = self._manager.set_peer_card(self._session_key, card_update, peer=peer)
             if result is None:
                 return tool_error("Failed to update peer card.")
@@ -833,6 +851,8 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
 
         if list_mode:
             return json.dumps({"conclusions": self._manager.list_conclusions(self._session_key, query=query or None, peer=peer)})
+        if refusal := self._bot_turn_write_refusal():
+            return refusal
         if delete_id:
             if self._manager.delete_conclusion(self._session_key, delete_id, peer=peer):
                 return json.dumps({"result": f"Conclusion {delete_id} deleted."})

plugins/memory/honcho/session.py

@@ -227,7 +227,7 @@ class HonchoSessionManager(SessionAuthMixin, SessionPeersMixin, SessionContextMi
         # identities to peerName for single-user deployments (see _resolve_user_peer_id).
         # Determine peer IDs — no lock needed (read-only, no shared state mutation). See #14984.
         user_peer_id = user_peer_id or self._resolve_user_peer_id(key)
-        assistant_peer_id = self._sanitize_id(self._config.ai_peer if self._config else "hermes-assistant")
+        assistant_peer_id = self.assistant_peer_id()
 
         # All expensive I/O outside the lock — Honcho's persistence is source of truth.
         honcho_session_id = self._sanitize_id(key)
@@ -250,9 +250,9 @@ class HonchoSessionManager(SessionAuthMixin, SessionPeersMixin, SessionContextMi
     # ----- Writes -----
 
     def _author_peer_for_session(self, honcho_session: Any, honcho_session_id: str, author_peer_id: str) -> Any:
-        """Return the author's peer, joining it to the session on first sight. A shared session's
-        roster is open (people and agents arrive later), so peers join when they first write; joins
-        are remembered per session so this costs one API call per author."""
+        """The author's peer, joined to the session the first time it writes.
+
+        Joins are remembered per session, so this costs one API call per author."""
         peer = self._get_or_create_peer(author_peer_id)
         with self._cache_lock:
             if author_peer_id in self._joined_author_peers.setdefault(honcho_session_id, set()):
@@ -262,7 +262,7 @@ class HonchoSessionManager(SessionAuthMixin, SessionPeersMixin, SessionContextMi
             config = SessionPeerConfig(observe_me=self._user_observe_me, observe_others=self._user_observe_others)
             honcho_session.add_peers([(peer, config)])
         except Exception as e:
-            # The write still lands under the right peer; only the membership (observe config) is missing.
+            # The write still lands under the right peer. Only the membership (observe config) is missing.
             logger.debug("Honcho author peer join failed for %s: %s", author_peer_id, e)
             return peer
         with self._cache_lock:

plugins/memory/honcho/session_peers.py

@@ -104,11 +104,10 @@ class SessionPeersMixin:
         return target_peer_id, None
 
     def _peer_id_for_runtime_id(self, runtime_id: str) -> str:
-        """Map one gateway runtime identity onto its Honcho peer ID with the same alias-then-prefix
-        order ``_resolve_user_peer_id`` applies, so an aliased account lands on its peer on any turn.
-        A ``bot:<profile>`` author maps onto the profile name: a cloned profile's aiPeer defaults to
-        that name, so a sender in the same workspace lands on its existing AI peer. Prefixes never
-        apply to bot ids."""
+        """Honcho peer ID for one runtime identity: alias first, then prefix, as at session init.
+
+        A ``bot:<profile>`` author maps onto the bare profile name because a cloned profile's aiPeer
+        defaults to that name."""
         aliases = getattr(self._config, "user_peer_aliases", {}) if self._config else {}
         if isinstance(aliases, dict):
             alias = aliases.get(runtime_id)
@@ -121,16 +120,22 @@ class SessionPeersMixin:
         prefix = prefix.strip() if isinstance(prefix, str) else ""
         return self._generated_runtime_peer_id(prefix, runtime_id) if prefix else self._sanitize_id(runtime_id)
 
-    def resolve_author_peer_id(self, key: str, author_id: str | None, author_name: str | None = None) -> str | None:
-        """Peer ID for the turn's author, or None to keep the session's peer: no author named, the
-        author IS the session's peer, or ``pinPeerName`` collapsing the operator's accounts. Bot authors
-        are never collapsed: pinned or not, a bot's words go under the bot's peer.
-        ``author_name`` is a display name (attacker-influenceable), so it never becomes a peer ID."""
+    def assistant_peer_id(self) -> str:
+        """This agent's own peer ID, as the session builder derives it from ``aiPeer``."""
+        return self._sanitize_id(self._cfg("ai_peer") or "hermes-assistant")
+
+    def resolve_author_peer_id(
+        self, key: str, author_id: str | None, author_name: str | None = None, *, is_bot: bool = False,
+    ) -> str | None:
+        """Peer ID for the turn's author. None keeps the session's own peer.
+
+        A bot author (``is_bot`` or a ``bot:`` id) always gets its own peer: the pin only collapses the
+        operator's accounts. ``author_name`` never becomes a peer ID because display names are
+        attacker-influenceable."""
         runtime_id = str(author_id).strip() if author_id else ""
         if not runtime_id:
             return None
-        # The pin collapses the operator's own accounts onto one peer; a bot is never one of those.
-        if runtime_id.startswith(BOT_AUTHOR_PREFIX):
+        if is_bot or runtime_id.startswith(BOT_AUTHOR_PREFIX):
             return self._peer_id_for_runtime_id(runtime_id)
         if self._config is not None and bool(getattr(self._config, "peer_name", None)) \
                 and getattr(self._config, "pin_peer_name", False) is True:

tests/honcho_plugin/test_a2a_sessions.py

@@ -42,7 +42,7 @@ class TestA2aRouting:
 
         _sync(provider, turn_author=BOT_AUTHOR)
 
-        provider._manager.get_or_create.assert_called_once_with("Bot-Chat:a2a:bot-coder", user_peer_id="coder")
+        provider._manager.get_or_create.assert_called_once_with(provider._a2a_session_key("bot:coder"), user_peer_id="coder")
         session = provider._manager.get_or_create.return_value
         roles = [c[0][0] for c in session.add_message.call_args_list]
         assert roles == ["user", "assistant"]
@@ -66,17 +66,17 @@ class TestA2aRouting:
         _sync(provider, turn_author=BOT_AUTHOR)
 
         keys = {c[0][0] for c in provider._manager.get_or_create.call_args_list}
-        assert keys == {"Bot-Chat:a2a:bot-coder"}
+        assert keys == {provider._a2a_session_key("bot:coder")}
 
     def test_two_bots_get_two_sessions(self):
         provider = _provider()
-        provider._manager.resolve_author_peer_id.side_effect = lambda key, author_id, name=None: author_id[4:]
+        provider._manager.resolve_author_peer_id.side_effect = lambda key, author_id, name=None, **kw: author_id[4:]
 
         _sync(provider, turn_author=BOT_AUTHOR)
         _sync(provider, turn_author={"id": "bot:writer", "name": "writer", "is_bot": True})
 
         keys = [c[0][0] for c in provider._manager.get_or_create.call_args_list]
-        assert keys == ["Bot-Chat:a2a:bot-coder", "Bot-Chat:a2a:bot-writer"]
+        assert keys == [provider._a2a_session_key("bot:coder"), provider._a2a_session_key("bot:writer")]
 
     def test_bot_author_without_an_id_is_skipped(self):
         provider = _provider()
@@ -85,14 +85,40 @@ class TestA2aRouting:
 
         provider._manager.get_or_create.assert_not_called()
 
-    def test_unresolvable_bot_peer_keeps_the_default_user_peer(self):
-        """When no author peer resolves, the a2a session still opens under the resolved peer."""
+    def test_bot_is_resolved_as_a_bot_whatever_its_id_looks_like(self):
+        """A platform bot carries a raw user id and the bot flag. The resolver must not treat it as a human."""
+        provider = _provider()
+        provider._manager.resolve_author_peer_id.return_value = "tg_5551234"
+
+        _sync(provider, turn_author={"id": "5551234", "name": "SomeBot", "is_bot": True})
+
+        provider._manager.resolve_author_peer_id.assert_called_once_with("Bot-Chat", "5551234", "SomeBot", is_bot=True)
+        provider._manager.get_or_create.assert_called_once_with(provider._a2a_session_key("5551234"), user_peer_id="tg_5551234")
+
+    def test_unresolvable_bot_peer_skips_the_write(self):
+        """A bot's words never land under the human's peer, so no peer means no write."""
         provider = _provider()
         provider._manager.resolve_author_peer_id.return_value = None
 
         _sync(provider, turn_author=BOT_AUTHOR)
 
-        provider._manager.get_or_create.assert_called_once_with("Bot-Chat:a2a:bot-coder")
+        provider._manager.get_or_create.assert_not_called()
+
+    def test_bot_colliding_with_this_agents_ai_peer_is_skipped(self):
+        """One peer cannot be both sides of a session."""
+        provider = _provider()
+        provider._manager.resolve_author_peer_id.return_value = "hermes"
+        provider._manager.assistant_peer_id.return_value = "hermes"
+
+        _sync(provider, turn_author=BOT_AUTHOR)
+
+        provider._manager.get_or_create.assert_not_called()
+
+    def test_ids_that_sanitize_alike_get_different_sessions(self):
+        provider = _provider()
+        keys = {provider._a2a_session_key(bot) for bot in ("bot:a.b", "bot:a-b", "bot:a_b", "bot:a:b")}
+        assert len(keys) == 4
+        assert all(key.startswith("Bot-Chat:a2a:bot-a") for key in keys)
 
     def test_long_session_key_stays_within_the_honcho_limit(self):
         provider = _provider()
@@ -106,6 +132,46 @@ class TestA2aRouting:
         assert key == provider._a2a_session_key("bot:coder")
 
 
+class TestToolWritesDuringBotTurn:
+    def _tools_provider(self, author: dict) -> HonchoMemoryProvider:
+        provider = _provider()
+        provider._turn_author = dict(author)
+        provider._manager.create_conclusion.return_value = True
+        provider._manager.delete_conclusion.return_value = True
+        provider._manager.set_peer_card.return_value = ["fact"]
+        provider._manager.list_conclusions.return_value = []
+        return provider
+
+    def test_conclude_and_delete_are_refused(self):
+        provider = self._tools_provider(BOT_AUTHOR)
+        assert "error" in json.loads(provider._tool_conclude({"conclusion": "likes tea"}))
+        assert "error" in json.loads(provider._tool_conclude({"delete_id": "c1"}))
+        provider._manager.create_conclusion.assert_not_called()
+        provider._manager.delete_conclusion.assert_not_called()
+
+    def test_listing_still_works(self):
+        provider = self._tools_provider(BOT_AUTHOR)
+        assert json.loads(provider._tool_conclude({"list": True})) == {"conclusions": []}
+
+    def test_profile_card_write_is_refused_but_read_works(self):
+        provider = self._tools_provider(BOT_AUTHOR)
+        assert "error" in json.loads(provider._tool_profile({"card": ["fact"]}))
+        provider._manager.set_peer_card.assert_not_called()
+        provider._manager.get_peer_card.return_value = ["fact"]
+        assert json.loads(provider._tool_profile({})) == {"result": ["fact"]}
+
+    def test_memory_mirror_is_skipped(self):
+        provider = self._tools_provider(BOT_AUTHOR)
+        provider.on_memory_write("add", "user", "likes tea")
+        assert provider._memwrite_thread is None
+        provider._manager.create_conclusion.assert_not_called()
+
+    def test_human_turn_writes_normally(self):
+        provider = self._tools_provider(HUMAN_AUTHOR)
+        assert json.loads(provider._tool_conclude({"conclusion": "likes tea"}))["result"].startswith("Conclusion saved")
+        assert "card" in json.loads(provider._tool_profile({"card": ["fact"]}))
+
+
 class TestFlagOff:
     def test_bot_turn_is_skipped(self):
         provider = _provider(a2a_sessions=False)
@@ -151,7 +217,7 @@ class TestManagerUserPeerOverride:
         mgr._get_or_create_peer = MagicMock(side_effect=lambda pid: MagicMock(name=f"peer:{pid}"))
         mgr._get_or_create_honcho_session = MagicMock(return_value=(MagicMock(), []))
 
-        session = mgr.get_or_create("Bot-Chat:a2a:bot-coder", user_peer_id="coder")
+        session = mgr.get_or_create("Bot-Chat:a2a:bot-coder-0123abcd", user_peer_id="coder")
 
         assert session.user_peer_id == "coder"
         assert session.assistant_peer_id == "hermes"

tests/honcho_plugin/test_turn_author_peers.py

@@ -100,10 +100,28 @@ class TestResolveAuthorPeerId:
         assert mgr.resolve_author_peer_id("telegram:dm1", "bot:") == "bot-"
 
     def test_pin_peer_name_does_not_collapse_bot_authors(self):
-        """The pin unifies the operator's accounts; a bot's words never land under the human's peer."""
+        """The pin unifies the operator's accounts. A bot's words never land under the human's peer."""
         mgr = _manager(_config(pin_peer_name=True), runtime_id="7654321")
         assert mgr.resolve_author_peer_id("telegram:dm1", "bot:coder") == "coder"
 
+    def test_platform_bot_gets_its_own_peer_even_when_pinned(self):
+        """A Telegram bot arrives with a raw user id and the bot flag, never a ``bot:`` id."""
+        mgr = _manager(_config(pin_peer_name=True, runtime_peer_prefix="tg_"), runtime_id="7654321")
+        assert mgr.resolve_author_peer_id("telegram:dm1", "5551234", "SomeBot", is_bot=True) == "tg_5551234"
+
+    def test_platform_bot_alias_wins(self):
+        mgr = _manager(_config(user_peer_aliases={"5551234": "coder"}), runtime_id="7654321")
+        assert mgr.resolve_author_peer_id("telegram:dm1", "5551234", is_bot=True) == "coder"
+
+    def test_platform_bot_matching_the_session_peer_still_gets_its_own_peer(self):
+        """The session's runtime id is the human's. A bot claiming it must not inherit the human's peer."""
+        mgr = _manager(_config(), runtime_id="7654321")
+        assert mgr.resolve_author_peer_id("telegram:dm1", "7654321", is_bot=True) == "7654321"
+
+    def test_assistant_peer_id_matches_the_session_builder(self):
+        assert _manager(_config(ai_peer="My Agent")).assistant_peer_id() == "My-Agent"
+        assert _manager(_config(ai_peer="")).assistant_peer_id() == "hermes-assistant"
+
     def test_display_name_never_becomes_a_peer_id(self):
         """Display names are attacker-influenceable on most platforms."""
         mgr = _manager(_config(), runtime_id="7654321")
@@ -248,7 +266,7 @@ class TestProviderReadsTheAuthor:
         assert all(c[1]["author_peer_id"] == "alice" for c in user_calls)
 
     def test_sync_turn_prefers_the_turn_author_keyword(self):
-        """The manager passes the author with the turn; the stash is only a fallback."""
+        """The manager passes the author with the turn. The stash is only a fallback."""
         provider = self._provider()
         provider._session_initialized = True
         provider._manager.get_or_create.return_value = MagicMock()