feat(bot-mode): message_agent tool — structured, Bot-Chat-only agent-to-agent DMs

Bot Mode agents now DM teammates through a real tool instead of
hand-assembled shell commands. message_agent(target, message) validates
the target against the live roster, applies the sender's attribution
prefix server-side, and delivers over the existing proven transports
(hermes -p ... --query-file for local teammates, hermes peer dm for
peer gateways) as a tracked background process with notify-on-complete
— fire-and-forget, the reply wakes the sender on a later turn.

Containment: the schema is injected per-turn ONLY into a bot's
canonical 'Bot Chat' session on Bot-Mode-managed installs (same gate as
the protocol section); it is never registered in the tool registry or
any toolset, and dispatch re-gates on the session title so a forged
call from any other session refuses. The gate is session-stable, so the
tool list stays byte-identical across turns (prompt-cache safe).

The protocol section is rewritten to teach the tool and now carries the
teammate roster WITH ROLES (Bot Mode title + profile description), so
bots know who does what before picking a recipient. Roles and a
protocol version salt join the capability fingerprint: existing eternal
Bot Chats adopt the v2 protocol + tool with one epoch refresh, and a
rename/description edit refreshes the roster on the next message.

agent/tool_executor.py

@@ -2066,6 +2066,31 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
                 agent._vprint(f"  {_get_cute_tool_message_impl('todo', function_args, tool_duration, result=function_result)}")
+        elif function_name == "message_agent":
+            # Bot Mode teammate DM (tools/bot_mode_dm.py) — injected, not
+            # registered: only a canonical Bot Chat session carries the
+            # schema, and the tool re-gates on the session title itself.
+            def _execute(next_args: dict) -> Any:
+                from tools.bot_mode_dm import message_agent_tool as _message_agent_tool
+                return _message_agent_tool(
+                    target=next_args.get("target", ""),
+                    message=next_args.get("message", ""),
+                    task_id=effective_task_id,
+                    agent=agent,
+                )
+            function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware(
+                agent,
+                function_name=function_name,
+                function_args=function_args,
+                effective_task_id=effective_task_id,
+                tool_call_id=getattr(tool_call, "id", "") or "",
+                execute=_execute,
+                scope_block=_ts_scope_block,
+                display_index=i,
+            ))
+            tool_duration = time.time() - tool_start_time
+            if agent._should_emit_quiet_tool_messages():
+                agent._vprint(f"  {_get_cute_tool_message_impl('message_agent', function_args, tool_duration, result=function_result)}")
         elif function_name == "session_search":
             def _execute(next_args: dict) -> Any:
                 session_db = agent._get_session_db_for_recall()

agent/turn_context.py

@@ -746,6 +746,19 @@ def build_turn_context(
 
     active_system_prompt = agent._cached_system_prompt
 
+    # Bot Mode DM tool — injected ONLY into a bot's canonical "Bot Chat"
+    # session on Bot-Mode-managed installs (same gate as the protocol
+    # section above). The gate is stable for a session's lifetime, so the
+    # tool list is byte-identical every turn: prompt-cache safe. Every
+    # other session (CLI, gateway chats, group-room member sessions, cron,
+    # subagents) fails the gate and never sees the schema.
+    try:
+        from tools.bot_mode_dm import ensure_message_agent_tool
+
+        ensure_message_agent_tool(agent)
+    except Exception:
+        logger.debug("message_agent injection skipped", exc_info=True)
+
     # Create the DB session row now that _cached_system_prompt is populated, so
     # the persisted snapshot is written non-NULL on the first turn (Issue
     # #45499). Idempotent: _ensure_db_session() no-ops once the row exists.

tests/hermes_cli/test_chat_query_file.py

@@ -61,15 +61,26 @@ def test_query_and_query_file_mutually_exclusive(tmp_path):
 
 
 def test_bot_mode_protocol_never_inlines_message_into_shell():
-    """The DM protocol must use --query-file / stdin, not -q "…" inlining."""
+    """The DM transport must use --query-file / stdin, not -q "…" inlining.
+
+    The transport moved from prompt-injected instructions (bot_mode_probe)
+    to the message_agent tool (bot_mode_dm) in Aug 2026 — the invariant now
+    holds on the tool's command builder, and the probe must no longer teach
+    any shellout at all.
+    """
     sys.path.insert(0, str(REPO))
     try:
         import importlib
 
+        dm = importlib.import_module("tools.bot_mode_dm")
+        src = Path(dm.__file__).read_text(encoding="utf-8")
         probe = importlib.import_module("tools.bot_mode_probe")
-        src = Path(probe.__file__).read_text(encoding="utf-8")
+        probe_src = Path(probe.__file__).read_text(encoding="utf-8")
     finally:
         sys.path.remove(str(REPO))
     assert "--query-file" in src
     assert '-q "Message from' not in src
     assert 'dm <peer>/<agent-name> "Message from' not in src
+    # The protocol section teaches the tool, never a hand-rolled shellout.
+    assert "message_agent" in probe_src
+    assert "--query-file /tmp/dm.txt" not in probe_src

tests/tools/test_bot_mode_dm.py

@@ -0,0 +1,296 @@
+"""Tests for tools/bot_mode_dm.py — the Bot-Chat-only ``message_agent`` tool.
+
+The containment contract is the headline here: the tool must exist ONLY in a
+canonical Bot Chat session on a Bot-Mode-managed install, and must refuse to
+deliver from anywhere else even if a schema leaks.
+"""
+
+import json
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from tools import bot_mode_dm, bot_mode_probe
+
+
+@pytest.fixture(autouse=True)
+def _fresh_probe_cache():
+    bot_mode_probe._reset_cache_for_tests()
+    yield
+    bot_mode_probe._reset_cache_for_tests()
+
+
+def _managed_home(tmp_path, *, teammates=("researcher",), peers=()) -> Path:
+    home = tmp_path / ".hermes"
+    home.mkdir(exist_ok=True)
+    for name in teammates:
+        d = home / "profiles" / name
+        d.mkdir(parents=True, exist_ok=True)
+        (d / "profile.yaml").write_text(
+            textwrap.dedent(
+                """\
+                description: teammate for tests
+                ui_meta:
+                  hermes-bots:
+                    shape: cloud
+                """
+            ),
+            encoding="utf-8",
+        )
+    if peers:
+        lines = ["bot_peers:"]
+        for peer in peers:
+            lines += [f"  {peer}:", f"    url: http://{peer}.lan:8377"]
+        (home / "config.yaml").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return home
+
+
+class _FakeDB:
+    def __init__(self, home: Path, title: str):
+        self.db_path = str(home / "state.db")
+        self._title = title
+
+    def get_session_title(self, _sid):
+        return self._title
+
+
+class _FakeAgent:
+    def __init__(self, home: Path, title: str = "Bot Chat"):
+        self._session_db = _FakeDB(home, title)
+        self.session_id = "sess-1"
+        self._session_title_hint = None
+        self._bot_mode_protocol = True
+        self.tools: list = []
+        self.valid_tool_names: set = set()
+
+
+# ── injection gate (leak containment) ────────────────────────────────────────
+
+
+def test_injects_only_into_bot_chat_on_managed_install(tmp_path):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Bot Chat")
+    assert bot_mode_dm.ensure_message_agent_tool(agent) is True
+    names = [t["function"]["name"] for t in agent.tools]
+    assert names == [bot_mode_dm.MESSAGE_AGENT_TOOL_NAME]
+    assert bot_mode_dm.MESSAGE_AGENT_TOOL_NAME in agent.valid_tool_names
+
+    # idempotent: second call adds nothing (byte-stable tool list per turn)
+    assert bot_mode_dm.ensure_message_agent_tool(agent) is True
+    assert len(agent.tools) == 1
+
+
+@pytest.mark.parametrize(
+    "title",
+    ["", "My research chat", "Group: room-abc123", "handoff-12ab34cd"],
+)
+def test_never_injects_outside_bot_chat(tmp_path, title):
+    """CLI sessions, ordinary chats, group-room member sessions: no tool."""
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title=title)
+    assert bot_mode_dm.ensure_message_agent_tool(agent) is False
+    assert agent.tools == []
+    assert agent.valid_tool_names == set()
+
+
+def test_never_injects_on_unmanaged_install(tmp_path):
+    """A 'Bot Chat'-titled session on a plain install stays tool-free."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    agent = _FakeAgent(home, title="Bot Chat")
+    assert bot_mode_dm.ensure_message_agent_tool(agent) is False
+    assert agent.tools == []
+
+
+def test_config_toggle_disables_injection(tmp_path):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Bot Chat")
+    agent._bot_mode_protocol = False
+    assert bot_mode_dm.ensure_message_agent_tool(agent) is False
+    assert agent.tools == []
+
+
+def test_schema_never_in_global_registry():
+    """message_agent must not be registered/toolset-reachable anywhere."""
+    from tools.registry import registry
+
+    assert bot_mode_dm.MESSAGE_AGENT_TOOL_NAME not in getattr(registry, "_tools", {})
+    import toolsets
+
+    for names in toolsets.TOOLSETS.values():
+        assert bot_mode_dm.MESSAGE_AGENT_TOOL_NAME not in names
+
+
+# ── dispatch gate (defense in depth) ─────────────────────────────────────────
+
+
+def test_tool_refuses_outside_bot_chat(tmp_path):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Ordinary chat")
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message="hi", agent=agent)
+    )
+    assert "error" in result
+    assert "Bot Chat" in result["error"]
+
+
+def test_tool_refuses_on_unmanaged_install(tmp_path):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    agent = _FakeAgent(home, title="Bot Chat")
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message="hi", agent=agent)
+    )
+    assert "error" in result
+
+
+# ── target validation ────────────────────────────────────────────────────────
+
+
+def test_unknown_target_lists_roster(tmp_path):
+    home = _managed_home(tmp_path, teammates=("researcher", "coder"))
+    agent = _FakeAgent(home, title="Bot Chat")
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="nosuchbot", message="hi", agent=agent)
+    )
+    assert "error" in result
+    assert set(result["teammates"]) == {"researcher", "coder"}
+
+
+def test_cannot_message_self(tmp_path):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Bot Chat")  # default profile
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="hermes", message="hi", agent=agent)
+    )
+    assert "error" in result
+    assert "yourself" in result["error"]
+
+
+def test_empty_and_oversized_message_rejected(tmp_path):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Bot Chat")
+    assert "error" in json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message="  ", agent=agent)
+    )
+    big = "x" * (bot_mode_dm.MESSAGE_MAX_CHARS + 1)
+    assert "error" in json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message=big, agent=agent)
+    )
+
+
+def test_unregistered_peer_rejected(tmp_path):
+    home = _managed_home(tmp_path, peers=("spark",))
+    agent = _FakeAgent(home, title="Bot Chat")
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="homelab/coder", message="hi", agent=agent)
+    )
+    assert "error" in result
+    assert result["peers"] == ["spark"]
+
+
+# ── delivery command shape ───────────────────────────────────────────────────
+
+
+def _capture_spawn(monkeypatch):
+    calls = []
+
+    def fake_terminal_tool(command, **kwargs):
+        calls.append({"command": command, **kwargs})
+        return json.dumps({"output": "Background process started", "session_id": "proc_test1234"})
+
+    import tools.terminal_tool as terminal_tool_module
+
+    monkeypatch.setattr(terminal_tool_module, "terminal_tool", fake_terminal_tool)
+    return calls
+
+
+def test_local_delivery_command_and_ack(tmp_path, monkeypatch):
+    calls = _capture_spawn(monkeypatch)
+    home = _managed_home(tmp_path, teammates=("researcher",))
+    agent = _FakeAgent(home, title="Bot Chat")
+
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(
+            target="@researcher",
+            message='status? give me the "final" numbers $(and this is not shell)',
+            agent=agent,
+        )
+    )
+    assert result["status"] == "sent"
+    assert result["to"] == "@researcher"
+    assert result["process_id"] == "proc_test1234"
+    assert "do NOT wait" in result["detail"]
+
+    assert len(calls) == 1
+    call = calls[0]
+    assert call["background"] is True
+    assert call["notify_on_complete"] is True
+    command = call["command"]
+    assert command.startswith("hermes -p researcher chat --in ~ -c \"Bot Chat\"")
+    assert "--query-file" in command
+    # message body rides the temp file, never the command line
+    assert "final" not in command
+    assert "$(" not in command
+
+    # attribution prefix applied server-side; body verbatim inside the file
+    dm_file = command.rsplit(" ", 1)[-1].strip("'")
+    content = Path(dm_file).read_text(encoding="utf-8")
+    assert content.startswith("Message from 🤖 hermes (@hermes): ")
+    assert '$(and this is not shell)' in content
+
+
+def test_peer_delivery_command(tmp_path, monkeypatch):
+    calls = _capture_spawn(monkeypatch)
+    home = _managed_home(tmp_path, peers=("spark",))
+    agent = _FakeAgent(home, title="Bot Chat")
+
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="spark/researcher", message="ping", agent=agent)
+    )
+    assert result["status"] == "sent"
+    assert "spark" in result["to"]
+    command = calls[0]["command"]
+    assert command.startswith("hermes peer dm spark/researcher < ")
+
+    # bare peer name targets the peer's main agent
+    result2 = json.loads(
+        bot_mode_dm.message_agent_tool(target="spark", message="ping", agent=agent)
+    )
+    assert result2["status"] == "sent"
+    assert calls[1]["command"].startswith("hermes peer dm spark < ")
+
+
+def test_named_profile_sender_prefix(tmp_path, monkeypatch):
+    """A named-profile bot signs with its own handle, not @hermes."""
+    calls = _capture_spawn(monkeypatch)
+    home = _managed_home(tmp_path, teammates=("researcher", "coder"))
+    profile_home = home / "profiles" / "coder"
+    agent = _FakeAgent(profile_home, title="Bot Chat")
+
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message="hi", agent=agent)
+    )
+    assert result["status"] == "sent"
+    dm_file = calls[0]["command"].rsplit(" ", 1)[-1].strip("'")
+    assert Path(dm_file).read_text(encoding="utf-8").startswith(
+        "Message from 🤖 coder (@coder): "
+    )
+
+
+def test_spawn_failure_reports_error(tmp_path, monkeypatch):
+    home = _managed_home(tmp_path)
+    agent = _FakeAgent(home, title="Bot Chat")
+
+    import tools.terminal_tool as terminal_tool_module
+
+    def boom(command, **kwargs):
+        raise RuntimeError("spawn failed")
+
+    monkeypatch.setattr(terminal_tool_module, "terminal_tool", boom)
+    result = json.loads(
+        bot_mode_dm.message_agent_tool(target="researcher", message="hi", agent=agent)
+    )
+    assert "error" in result
+    assert "could not be started" in result["error"]

tests/tools/test_bot_mode_probe.py

@@ -51,8 +51,8 @@ def test_emits_for_default_when_any_profile_is_managed(tmp_path):
     # default's callable alias is @hermes, never @default
     assert "@hermes" in section
     assert "@default" not in section
-    assert "`researcher`" in section
-    assert "hermes profile list" in section
+    assert "@researcher" in section
+    assert "message_agent" in section
 
 
 def test_emits_for_named_profile_with_own_handle(tmp_path):
@@ -62,9 +62,36 @@ def test_emits_for_named_profile_with_own_handle(tmp_path):
 
     section = bot_mode_probe.get_bot_mode_protocol_section(profile_dir)
     assert "@coder" in section
-    # teammate list excludes self, includes default
-    assert "`default`" in section
-    assert "`coder`" not in section.split("Teammates at session start:")[1]
+    # teammate roster excludes self, includes default (as @hermes)
+    roster_block = section.split("Your teammates")[1]
+    assert "`@hermes`" in roster_block
+    assert "`@coder`" not in roster_block
+
+
+def test_roster_lines_carry_roles(tmp_path):
+    """Bots must know WHO to message: the roster carries title/description."""
+    import textwrap as _tw
+
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    d = home / "profiles" / "researcher"
+    d.mkdir(parents=True)
+    (d / "profile.yaml").write_text(
+        _tw.dedent(
+            """\
+            description: Deep research and literature review
+            ui_meta:
+              hermes-bots:
+                title: Research Buddy
+            """
+        ),
+        encoding="utf-8",
+    )
+
+    section = bot_mode_probe.get_bot_mode_protocol_section(home)
+    assert "`@researcher`" in section
+    assert "Research Buddy" in section
+    assert "Deep research and literature review" in section
 
 
 def test_silent_when_soul_already_carries_protocol(tmp_path):
@@ -233,7 +260,8 @@ def test_peer_paragraph_lists_registered_peers(tmp_path):
     )
 
     section = bot_mode_probe.get_bot_mode_protocol_section(home)
-    assert "hermes peer dm" in section
+    assert "message_agent" in section
+    assert '"<peer>/<agent-name>"' in section
     assert "`homelab`" in section and "`spark`" in section
     assert "hermes peer list" in section
 

tools/bot_mode_dm.py

@@ -0,0 +1,384 @@
+"""Bot Mode agent-to-agent DM tool — ``message_agent``.
+
+A structured, Bot-Chat-only tool that lets a Bot Mode agent message a
+teammate agent (another Hermes profile on this install, or an agent on a
+registered peer gateway) WITHOUT hand-assembling shell commands.
+
+Why this exists (Aug 2026): the Bot Mode teammate protocol taught agents to
+DM each other via a prompt-injected ``hermes -p <bot> chat ...`` shellout.
+That transport works, but the *invocation* was fragile — quoting traps
+(#91339/#91304), temp-file choreography, dead-profile races — and the
+Desktop's remote-mention path forwarded raw user text verbatim (#91397).
+``message_agent`` replaces the invocation with a real tool call: the message
+is a parameter, the target is validated against the live roster, the
+attribution prefix is applied server-side, and the reply arrives through the
+existing background-process notification path (fire-and-forget, never
+blocks the sender's turn).
+
+Containment contract (MUST hold — reviewers check all three):
+- The tool schema is injected ONLY into a bot's canonical "Bot Chat"
+  session on Bot-Mode-managed installs — the exact same gate as the
+  protocol section in ``tools/bot_mode_probe.py``. It is NOT registered in
+  the global tool registry, is NOT part of any toolset, and never appears
+  in CLI sessions, ordinary gateway chats, group-room member sessions
+  (titled "Group: …"), cron agents, or subagents.
+- Dispatch is title-gated again at execution time (defense in depth): a
+  forged call from a session that shouldn't have the tool returns a
+  structured error instead of delivering.
+- Everything here is additive. The legacy protocol transports
+  (``hermes -p`` / ``hermes peer dm``) keep working for older prompts.
+
+The transports themselves are unchanged and proven:
+- local teammate  → ``hermes -p <name> chat --in ~ -c "Bot Chat"
+  --create-if-missing -Q --query-file <tmp>`` (one turn, reply on stdout)
+- peer teammate   → ``hermes peer dm <peer>[/<name>] < <tmp>``
+
+Both run through ``terminal_tool(background=True, notify_on_complete=True)``
+so the reply lands as a completion notification on the sender's NEXT turn —
+the same wake shape every Bot Mode agent already knows.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import shlex
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+MESSAGE_AGENT_TOOL_NAME = "message_agent"
+
+# Message body cap — generous for real work products, small enough that a
+# runaway paste can't turn one DM into a context bomb on the recipient.
+MESSAGE_MAX_CHARS = 16000
+
+_PEER_TARGET_RE = re.compile(r"^([a-z0-9][a-z0-9_-]{0,63})/([a-zA-Z0-9][a-zA-Z0-9_-]{0,63})$")
+_LOCAL_TARGET_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$")
+
+
+def message_agent_tool_schema() -> dict:
+    """OpenAI-format schema for ``message_agent`` (injected, not registered)."""
+    return {
+        "type": "function",
+        "function": {
+            "name": MESSAGE_AGENT_TOOL_NAME,
+            "description": (
+                "Send a message to ANOTHER agent (teammate) on this install, or to an "
+                "agent on a registered peer gateway. This is FIRE-AND-FORGET and "
+                "asynchronous, like texting: it validates the target against the live "
+                "roster, delivers your message into that agent's own Bot Chat with your "
+                "attribution automatically prefixed, and returns immediately with a "
+                "delivery acknowledgement. It does NOT return their reply and you must "
+                "not wait or poll for one — send it, finish your turn, and the reply "
+                "arrives later as a background-process completion notification that "
+                "wakes you. COMPOSE the message yourself: write what YOU want to say to "
+                "that agent (lead with the point; include the concrete ask or result). "
+                "Never paste the user's words verbatim — paraphrase the actionable "
+                "substance, and keep private 1:1 chat content private. Message one "
+                "clearly relevant teammate when it genuinely helps the user's goal; "
+                "don't fan out to several agents unless the user explicitly asked. "
+                "Use the teammate roster in your system prompt (names + roles) to pick "
+                "the right recipient; targets: a teammate name (e.g. 'researcher'), or "
+                "'<peer>/<agent>' for an agent on a registered peer gateway "
+                "(e.g. 'spark/researcher', or just '<peer>' for the peer's main agent)."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "target": {
+                        "type": "string",
+                        "description": (
+                            "Who to message: a teammate profile name from your roster "
+                            "('researcher', 'hermes' for the default agent), or "
+                            "'<peer>' / '<peer>/<agent>' for a registered peer gateway."
+                        ),
+                    },
+                    "message": {
+                        "type": "string",
+                        "description": (
+                            "The message YOU composed for that agent (max "
+                            f"{MESSAGE_MAX_CHARS} chars). Do not include the "
+                            "'Message from …' prefix — it is added automatically."
+                        ),
+                    },
+                },
+                "required": ["target", "message"],
+            },
+        },
+    }
+
+
+def ensure_message_agent_tool(agent: Any) -> bool:
+    """Inject the ``message_agent`` schema into a Bot Chat agent's tool list.
+
+    Called once per turn from the conversation loop. Idempotent and
+    deterministic for the life of a session: the gate (canonical Bot Chat
+    title on a Bot-Mode-managed install) is stable from the session's first
+    turn, so the tool list is byte-identical across turns — prompt-cache
+    safe. Every non-Bot-Chat session fails the gate on every turn and never
+    sees the schema. Never raises.
+    """
+    try:
+        if not getattr(agent, "_bot_mode_protocol", True):
+            return False
+        tools = getattr(agent, "tools", None)
+        if tools:
+            for tool in tools:
+                if (
+                    isinstance(tool, dict)
+                    and tool.get("function", {}).get("name") == MESSAGE_AGENT_TOOL_NAME
+                ):
+                    return True
+        from tools.bot_mode_probe import BOT_CHAT_TITLE, get_bot_mode_protocol_section
+
+        if _session_title(agent) != BOT_CHAT_TITLE:
+            return False
+        if not get_bot_mode_protocol_section(_agent_home(agent)):
+            return False
+        if agent.tools is None:
+            agent.tools = []
+        agent.tools.append(message_agent_tool_schema())
+        valid = getattr(agent, "valid_tool_names", None)
+        if isinstance(valid, set):
+            valid.add(MESSAGE_AGENT_TOOL_NAME)
+        return True
+    except Exception:  # pragma: no cover — must never break a turn
+        logger.debug("ensure_message_agent_tool failed", exc_info=True)
+        return False
+
+
+# ── roster resolution ────────────────────────────────────────────────────────
+
+
+def _hermes_root(home: Path) -> Path:
+    if home.parent.name == "profiles":
+        return home.parent.parent
+    return home
+
+
+def _self_profile_name(home: Path) -> str:
+    if home.parent.name == "profiles":
+        return home.name
+    return "default"
+
+
+def _local_roster(root: Path) -> list[str]:
+    """Profile names on this install: default + every named profile."""
+    names = ["default"]
+    try:
+        profiles = root / "profiles"
+        if profiles.is_dir():
+            for child in sorted(profiles.iterdir()):
+                if child.is_dir():
+                    names.append(child.name)
+    except Exception:
+        pass
+    return names
+
+
+def _peers(root: Path) -> list[str]:
+    try:
+        from tools.bot_mode_probe import _peers as _probe_peers
+
+        return _probe_peers(root)
+    except Exception:
+        return []
+
+
+def _handle(name: str) -> str:
+    return "hermes" if name == "default" else name
+
+
+def _resolve_local_name(target: str, roster: list[str]) -> Optional[str]:
+    """Map a target handle to a profile name ('hermes' → 'default')."""
+    want = target.strip()
+    if not want:
+        return None
+    if want.lower() == "hermes":
+        return "default" if "default" in roster else None
+    for name in roster:
+        if name.lower() == want.lower():
+            return name
+    return None
+
+
+# ── the tool ─────────────────────────────────────────────────────────────────
+
+
+def _err(message: str, *, roster: list[str] | None = None, peers: list[str] | None = None) -> str:
+    payload: dict[str, Any] = {"error": message}
+    if roster is not None:
+        payload["teammates"] = roster
+    if peers is not None:
+        payload["peers"] = peers
+    return json.dumps(payload)
+
+
+def message_agent_tool(
+    target: str = "",
+    message: str = "",
+    task_id: Optional[str] = None,
+    agent: Any = None,
+) -> str:
+    """Deliver ``message`` to ``target``'s Bot Chat. Returns a JSON ack/error.
+
+    ``agent`` is the calling AIAgent (threaded by the executor) — used for
+    the Bot Chat gate, the sender identity, and the session key so the
+    spawned transport is tracked against the right session.
+    """
+    # ── defense-in-depth gate: only a canonical Bot Chat may deliver ──
+    home = _agent_home(agent)
+    try:
+        from tools.bot_mode_probe import BOT_CHAT_TITLE, get_bot_mode_protocol_section
+
+        title = _session_title(agent)
+        if title != BOT_CHAT_TITLE:
+            return _err(
+                "message_agent is only available in a Bot Mode 'Bot Chat' session. "
+                "This session is not one; do not retry."
+            )
+        if not get_bot_mode_protocol_section(home):
+            return _err(
+                "This install is not Bot-Mode-managed (no bot roster); "
+                "message_agent is unavailable. Do not retry."
+            )
+    except Exception as exc:  # pragma: no cover — defensive
+        return _err(f"Bot Mode gate check failed: {exc}")
+
+    root = _hermes_root(Path(home))
+    me = _self_profile_name(Path(home))
+    roster = _local_roster(root)
+    peers = _peers(root)
+    teammates = [_handle(n) for n in roster if n != me]
+
+    body = str(message or "").strip()
+    if not body:
+        return _err("message is required — compose what you want to say to that agent.")
+    if len(body) > MESSAGE_MAX_CHARS:
+        return _err(
+            f"message too long ({len(body)} chars > {MESSAGE_MAX_CHARS}). "
+            "Send the essentials; share large content as a file path instead."
+        )
+
+    raw_target = str(target or "").strip().lstrip("@")
+    if not raw_target:
+        return _err("target is required.", roster=teammates, peers=peers)
+
+    sender_handle = _handle(me)
+    prefix = f"Message from 🤖 {sender_handle} (@{sender_handle}): "
+
+    # ── peer target: '<peer>/<agent>' or a bare registered peer name ──
+    peer_match = _PEER_TARGET_RE.match(raw_target)
+    bare_peer = raw_target.lower() if raw_target.lower() in peers else None
+    if peer_match or bare_peer:
+        peer_name = peer_match.group(1) if peer_match else bare_peer
+        peer_profile = peer_match.group(2) if peer_match else None
+        if peer_name not in peers:
+            return _err(
+                f"No registered peer named '{peer_name}'.", roster=teammates, peers=peers
+            )
+        dm_target = f"{peer_name}/{peer_profile}" if peer_profile else peer_name
+        command = f"hermes peer dm {shlex.quote(dm_target)} < {shlex.quote(_write_dm_file(prefix + body))}"
+        label = f"@{peer_profile or peer_name} on peer '{peer_name}'"
+        return _spawn_delivery(command, label, task_id=task_id, agent=agent)
+
+    # ── local teammate ──
+    if not _LOCAL_TARGET_RE.match(raw_target):
+        return _err(f"Invalid target: {raw_target!r}.", roster=teammates, peers=peers)
+    resolved = _resolve_local_name(raw_target, roster)
+    if resolved is None:
+        return _err(
+            f"No teammate named '{raw_target}' on this install. "
+            "Pick a name from the roster (roles are listed in your system prompt).",
+            roster=teammates,
+            peers=peers,
+        )
+    if resolved == me:
+        return _err("You can't message yourself. Pick a teammate from the roster.")
+
+    dm_file = _write_dm_file(prefix + body)
+    command = (
+        f"hermes -p {shlex.quote(resolved)} chat --in ~ -c \"Bot Chat\" "
+        f"--create-if-missing -Q --query-file {shlex.quote(dm_file)}"
+    )
+    return _spawn_delivery(command, f"@{_handle(resolved)}", task_id=task_id, agent=agent)
+
+
+def _write_dm_file(content: str) -> str:
+    """The message rides a temp file — never inline shell text."""
+    fd, path = tempfile.mkstemp(prefix="hermes-dm-", suffix=".txt", text=True)
+    with os.fdopen(fd, "w", encoding="utf-8") as f:
+        f.write(content)
+    return path
+
+
+def _spawn_delivery(command: str, label: str, *, task_id: Optional[str], agent: Any) -> str:
+    """Run the delivery command tracked + background, notify on completion."""
+    try:
+        from tools.terminal_tool import terminal_tool
+
+        raw = terminal_tool(
+            command,
+            background=True,
+            notify_on_complete=True,
+            task_id=task_id,
+        )
+        try:
+            parsed = json.loads(raw)
+        except (ValueError, TypeError):
+            parsed = {}
+        proc_id = parsed.get("session_id") or ""
+        if parsed.get("error"):
+            return _err(f"Delivery to {label} failed to start: {parsed['error']}")
+        return json.dumps(
+            {
+                "status": "sent",
+                "to": label,
+                "detail": (
+                    f"Message dispatched to {label}. This is asynchronous — do NOT wait "
+                    "or poll. Finish your turn now; when the delivery completes, its "
+                    "notification carries the reply — relay it then, attributed to "
+                    "that agent."
+                ),
+                **({"process_id": proc_id} if proc_id else {}),
+                "sent_at": int(time.time()),
+            }
+        )
+    except Exception as exc:
+        logger.error("message_agent delivery spawn failed: %s", exc, exc_info=True)
+        return _err(f"Delivery to {label} could not be started: {exc}")
+
+
+# ── agent-context helpers (mirror system_prompt.py's resolution) ─────────────
+
+
+def _agent_home(agent: Any) -> str:
+    """The calling agent's OWN home (session-db derived), not ambient env."""
+    try:
+        sdb = getattr(agent, "_session_db", None)
+        db_path = getattr(sdb, "db_path", None)
+        if db_path:
+            return str(Path(db_path).parent)
+    except Exception:
+        pass
+    return os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes")
+
+
+def _session_title(agent: Any) -> str:
+    title = str(getattr(agent, "_session_title_hint", "") or "").strip()
+    if title:
+        return title
+    try:
+        sdb = getattr(agent, "_session_db", None)
+        sid = getattr(agent, "session_id", None)
+        if sdb and sid:
+            return str(sdb.get_session_title(sid) or "").strip()
+    except Exception:
+        pass
+    return ""

tools/bot_mode_probe.py

@@ -105,6 +105,51 @@ def _handle(name: str) -> str:
     return "hermes" if name == "default" else name
 
 
+def _profile_role(profile_dir: Path) -> str:
+    """A teammate's role line: Bot Mode title, else profile description.
+
+    The ui_meta['hermes-bots'].title is the name the user gave the bot in
+    Bot Mode; profile.yaml's description is the profile's stated purpose.
+    Either one tells a teammate WHO to message for a given job. Bounded and
+    single-line; empty when neither exists. Never raises.
+    """
+    meta = profile_dir / "profile.yaml"
+    try:
+        if not meta.is_file():
+            return ""
+        raw = meta.read_text(encoding="utf-8", errors="replace")
+        import yaml
+
+        data = yaml.safe_load(raw)
+        if not isinstance(data, dict):
+            return ""
+        parts = []
+        ui_meta = data.get("ui_meta")
+        if isinstance(ui_meta, dict) and isinstance(ui_meta.get("hermes-bots"), dict):
+            title = str(ui_meta["hermes-bots"].get("title") or "").strip()
+            if title:
+                parts.append(title)
+        description = str(data.get("description") or "").strip()
+        if description:
+            parts.append(description)
+        line = " — ".join(parts)
+        return " ".join(line.split())[:160]
+    except Exception:
+        return ""
+
+
+def _roster_lines(root: Path, me: str) -> list[str]:
+    """One '- `@handle` — role' line per teammate (excluding ``me``)."""
+    lines = []
+    for name, profile_dir in _roster(root):
+        if name == me:
+            continue
+        role = _profile_role(profile_dir)
+        handle = _handle(name)
+        lines.append(f"- `@{handle}`" + (f" — {role}" if role else ""))
+    return lines
+
+
 def _peers(root: Path) -> list[str]:
     """Registered peer gateway names (``hermes peer``), for the protocol text.
 
@@ -137,15 +182,10 @@ def _peer_paragraph(root: Path) -> str:
     listed = ", ".join(f"`{p}`" for p in peers)
     return (
         "\n\nTeammates on OTHER machines: this install also has peer gateways "
-        f"registered ({listed}). Message an agent on a peer the same way — write "
-        "the message to a temp file first, then pipe it on stdin (same terminal-"
-        "tool pattern: background=true, notify_on_complete=true; the reply prints "
-        "on stdout when it completes):\n"
-        "```\n"
-        "hermes peer dm <peer>/<agent-name> < /tmp/dm.txt\n"
-        "```\n"
-        "Use `<peer>` alone for the peer's main agent. Run `hermes peer list` "
-        "for the live peer list."
+        f"registered ({listed}). Message an agent on a peer the same way — "
+        'message_agent with target "<peer>/<agent-name>" (or "<peer>" alone '
+        "for the peer's main agent). Run `hermes peer list` for the live "
+        "peer list."
     )
 
 
@@ -164,26 +204,32 @@ def _build_section(home: Path) -> str:
         return ""
 
     handle = _handle(me)
-    teammates = ", ".join(f"`{n}`" for n, _d in roster if n != me) or "(none yet)"
+    roster_block = "\n".join(_roster_lines(root, me)) or "- (no teammates yet)"
 
     return (
         f"{_PROTOCOL_HEADING}\n"
         "This install runs Bot Mode: each Hermes profile is an agent teammate with "
-        'one canonical "Bot Chat" conversation. To message a teammate: write the '
-        "message to a temp file with the file tool FIRST (never inline it into the "
-        "command — quotes truncate it and $( ) would execute), then run on the "
-        "terminal tool (background=true, notify_on_complete=true) and finish your "
-        "turn — the reply arrives later as a new message:\n"
-        "```\n"
-        f'hermes -p <agent-name> chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file /tmp/dm.txt\n'
-        "```\n"
-        f'The file must open with the "Message from 🤖 {handle} (@{handle}):" prefix so they '
-        "know who is talking. When YOU receive a message with that prefix, you are "
-        "being messaged by a teammate agent — address them (not the user) and reply "
-        "concisely. When the user says \"ask <name>\" or \"tell <name> ...\", that is a "
-        "handoff: message that agent, wait for the reply, and report back, saying "
-        "which agent it came from. Run `hermes profile list` for the LIVE teammate "
-        f"list before a handoff. Teammates at session start: {teammates}."
+        'one canonical "Bot Chat" conversation, and you have the `message_agent` '
+        "tool to DM any of them. It is FIRE-AND-FORGET: it delivers your message "
+        "with your attribution prefixed automatically and returns an acknowledgement "
+        "immediately — it never returns the reply. Send it, finish your turn, and "
+        "the reply arrives later as a background-process completion notification "
+        "that wakes you; relay it to the user then, attributed to that agent. "
+        "COMPOSE every message yourself — say what YOU need from that agent; never "
+        "forward the user's words verbatim, and never reveal private 1:1 chat "
+        "content. When the user says \"ask <name>\" or \"tell <name> ...\", that is "
+        "a handoff: pick the right teammate from the roster below, message them "
+        "with message_agent, and report back naming which agent replied. Message "
+        "ONE clearly relevant teammate; don't fan out to several unless the user "
+        "explicitly asked.\n"
+        f'When YOU receive a "Message from 🤖 <name> (@<handle>):" message, a '
+        "teammate agent is talking to you (not the user): address them, reply "
+        "concisely via message_agent to their handle, and if it is a pure FYI "
+        "with nothing to add, staying silent is fine — never ping-pong "
+        "acknowledgements.\n"
+        f"You are `@{handle}`. Your teammates (live roster; roles from their "
+        "profiles):\n"
+        f"{roster_block}"
         + _peer_paragraph(root)
     )
 
@@ -275,8 +321,18 @@ def capability_fingerprint(home: str | os.PathLike | None = None) -> str:
     try:
         root = _hermes_root(resolved)
         surface["roster"] = sorted(n for n, d in _roster(root) if _is_bot_managed(d))
+        # Roles are part of the messaging surface: renaming a bot or editing
+        # a profile description must refresh eternal Bot Chat prompts so the
+        # roster block teammates pick recipients from stays current.
+        surface["roster_roles"] = sorted(
+            f"{n}:{_profile_role(d)}" for n, d in _roster(root)
+        )
     except Exception:
         surface["roster"] = []
+    # Protocol-text version salt: bumping this refreshes every eternal Bot
+    # Chat prompt ONCE so existing bots adopt a new protocol section (e.g.
+    # the v2 message_agent tool replacing the shellout instructions).
+    surface["protocol_version"] = 2
     try:
         # Peer gateways are part of the messaging surface: registering one
         # must refresh eternal Bot Chat prompts so the cross-machine DM

website/docs/user-guide/bot-mode.md

@@ -97,7 +97,7 @@ Bots message each other with attribution, and you can hand work off from any cha
 - **@mentions** — type `@researcher have a look at this` in any chat and the active Bot hands the message off, waits for the reply, and reports back. Mention names are validated against the live roster, so an email address or an unknown `@` passes through untouched.
 - **Renamed Bots keep their tags in sync** — give a Bot a friendly name (the pencil in its chat header, or `hermes profile rename`) and it becomes taggable by that name: a Bot titled *Research Buddy* answers to `@research-buddy` (and `@researchbuddy`), in regular chats and in group rooms alike. The composer's `@` autocomplete offers the renamed tag and also matches when you type the old profile name, which keeps resolving too.
 - **@mentions across machines** — mentioning a Bot that lives on another registered connection (use its `@name-device` handle when names collide) delivers over the Connections registry in the background: the active Bot stays on this device, the desktop routes the message to the recipient's machine, and the reply is relayed back attributed to that agent. Your window's gateway never switches.
-- **Direct messages** — a Bot reaches a teammate's Bot Chat through the standard CLI: it writes the message to a temp file (opening with the `Message from 🤖 <sender> (@<sender>):` prefix), then runs `hermes -p <bot> chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file <file>`. The file transport means nothing is shell-interpreted — quotes, `$(...)`, and backticks in the message arrive verbatim. The receiving Bot sees the message the next time it runs and knows how to reply, because the messaging protocol is part of its Bot Chat system prompt.
+- **Direct messages** — every Bot Chat carries the `message_agent` tool: a Bot messages a teammate by calling `message_agent(target="researcher", message="…")`. The tool validates the target against the live roster, prefixes the sender's `Message from 🤖 <sender> (@<sender>):` attribution automatically, and delivers into the teammate's canonical Bot Chat. Delivery is **fire-and-forget**: the sender gets an acknowledgement, finishes its turn, and the reply arrives later as a background completion notification. The message travels as a real parameter (nothing shell-interpreted — quotes, `$(...)`, and backticks arrive verbatim), and the Bot composes its own message rather than forwarding your words. The teammate roster — names **and roles** from each profile's title/description — is part of every Bot Chat's system prompt, so Bots know who does what before choosing a recipient. The tool exists **only** in canonical Bot Chat sessions on Bot-Mode-managed installs; regular chats, group-room member sessions, and CLI sessions never see it.
 
 The backend teaches each Bot's canonical Bot Chat session the messaging protocol automatically at prompt-build time — including when a teammate opens it headlessly from the CLI. Only the canonical Bot Chat gets the protocol section; your regular sessions and your SOUL.md stay untouched. This is controlled by `agent.bot_mode_protocol` in `config.yaml` (default: on):
 
@@ -123,7 +123,7 @@ hermes peer dm spark/researcher < /tmp/dm.txt   # named profile on a multiplexed
 
 `hermes peer dm` delivers into the remote agent's canonical Bot Chat over the peer's existing API server, runs one agent turn there, and prints the reply on stdout — the exact cross-machine twin of the local `hermes -p <bot> chat` command.
 
-Once a peer is registered, the messaging protocol taught to every Bot Chat (`agent.bot_mode_protocol`) automatically includes the peer roster and the `hermes peer dm` pattern — so **your bots learn on their own** that teammates exist on other machines and how to reach them. Registering or removing a peer refreshes each Bot Chat's protocol on its next message (capability epoch).
+Once a peer is registered, the messaging protocol taught to every Bot Chat (`agent.bot_mode_protocol`) automatically includes the peer roster, and `message_agent` accepts peer targets directly — `message_agent(target="spark/researcher", …)`, or `target="spark"` for the peer's main agent — so **your bots learn on their own** that teammates exist on other machines and how to reach them. Registering or removing a peer refreshes each Bot Chat's protocol on its next message (capability epoch).
 
 Requirements: the peer machine runs the `api_server` gateway platform with a strong `API_SERVER_KEY`; reachability is your network's business (LAN, Tailscale, VPN). The key is a credential and lives in `~/.hermes/.env` as `HERMES_PEER_<NAME>_KEY`; peer names/URLs live in `config.yaml` under `bot_peers`.