fix(gateway): end bot-to-bot exchanges: no runtime notices to bots, a runtime hop cap, replies need no answer

Widens #137037 from Mattermost to the whole class (#137027):

- Runtime notices never reach a bot peer: every busy-path reply (ack, redirect,
  drain), lifecycle/status and credits notices, tool progress, interim and thinking
  bubbles, the long-running heartbeat, the auto-reset notice, and the runtime failure
  copy ("not processed", retry hints, generic failure apology) on a failed or empty
  turn. The model's own reply is still delivered.
- Chain rules in gateway.bot_loop_guard (config.yaml): a human message starts a chain,
  each bot message after it is one hop, max_hops (default 4) caps them, a bot repeating
  the same text in one chain is dropped, and with bot_replies_expect_reply false (the
  default) a bot answering the reply this bot just gave it starts no turn. A bot
  message also counts as not expecting a reply, so a silence marker ends it quietly.
- Bot authorship everywhere it is knowable: a generic per-adapter bot_users opt-in
  (config.yaml platform extra) for platforms whose events carry no bot flag,
  Mattermost props.from_bot, QQ guild author.bot.
- Bot Mode message_agent: agent/bot_chain.py owns a per-session hop count carried in
  the delivery's author envelope (runtime-written, parse_turn_author keeps it), refuses
  sends past the same max_hops and identical repeated asks, sends a reply to the bot
  that messaged this turn notify-only (expects_reply false, no wake), and refuses a
  message back to a bot whose reply said it needs no answer. tools/bot_mode_dm.py gets
  only the gate call and the wake flag (LANE dm owns that file; same wake names as #144789).

agent/bot_chain.py

@@ -0,0 +1,99 @@
+"""Bot-to-bot chains in Bot Mode: a runtime-owned hop count, identical-ask dedupe and the reply rule.
+
+A human message starts a chain. Each ``message_agent`` send inside it is one hop; the hop number rides
+in the delivery's author envelope (``hops``), which only the runtime writes, so the model cannot reset
+it. Past ``gateway.bot_loop_guard.max_hops`` (config.yaml, the same cap the messaging gateway applies)
+a send is refused and nothing is delivered. The same message to the same teammate twice in one chain
+is refused as a duplicate. A message to the bot whose message started this turn is a reply: it goes
+notify-only (``expects_reply`` false in the envelope, no wake for its answer), and a turn started by
+such a reply cannot message that bot back. Each side also counts the bot turns it runs, so a lane that
+carries no envelope (Desktop relay) is still capped (#137027).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import threading
+import time
+from typing import Any, Optional
+
+# Keyed by session id, not held on the agent: a gateway may rebuild the agent between turns, and a
+# lost count would hand the chain a fresh budget. Expired chains are pruned on every write.
+_CHAINS: dict[str, dict] = {}
+_LOCK = threading.Lock()
+
+
+def _settings():
+    from gateway.bot_loop_guard import load_settings
+    return load_settings()
+
+
+def _hops(author: Optional[dict]) -> int:
+    value = (author or {}).get("hops")
+    return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0
+
+
+def _profile_part(identity: str) -> str:
+    """``bot:<profile>`` / ``bot:<origin>/<profile>`` / a target ``peer/agent`` or ``agent@conn`` -> agent."""
+    text = str(identity or "").strip().lower().removeprefix("bot:").lstrip("@")
+    return text.rsplit("/", 1)[-1].split("@", 1)[0]
+
+
+def on_turn_start(agent: Any, turn_author: Optional[dict], display_kind: Optional[str]) -> None:
+    """Set the hop this session's turn runs at. A bot-authored turn runs at the hop its envelope names
+    (or one past this side's own count, whichever is higher); a completion wake after this bot sent a
+    message carries the answer, which is one more hop; a person's message ends the chain."""
+    from gateway.response_filters import is_machinery_display_kind
+    key, ttl = _key(agent), _settings().cooldown_seconds
+    now = time.monotonic()
+    with _LOCK:
+        for stale in [k for k, v in _CHAINS.items() if now - v["updated"] > ttl]:
+            del _CHAINS[stale]
+        state = _CHAINS.get(key)
+        if isinstance(turn_author, dict) and turn_author.get("is_bot"):
+            state = _CHAINS.setdefault(key, _new_state(now))
+            state["hops"] = max(state["hops"] + 1, _hops(turn_author))
+        elif state is not None and is_machinery_display_kind(display_kind):
+            if state["sent"] > state["hops"]:
+                state["hops"] = state["sent"] + 1
+        elif state is not None:
+            del _CHAINS[key]  # a person spoke: a new chain starts with their message
+            return
+        if state is not None:
+            state["updated"] = now
+
+
+def _new_state(now: float) -> dict:
+    return {"hops": 0, "sent": 0, "asks": set(), "updated": now}
+
+
+def _key(agent: Any) -> str:
+    return str(getattr(agent, "session_id", "") or id(agent))
+
+
+def admit_send(agent: Any, target: str, body: str) -> tuple[Optional[str], dict]:
+    """Check one ``message_agent`` send. Returns ``(refusal, envelope)``: a refusal sentence for the
+    model (nothing is delivered), else the fields the delivery's author envelope carries."""
+    author = getattr(agent, "_turn_author", None) or {}
+    replying = bool(author.get("is_bot")) and _profile_part(author.get("id") or author.get("name") or "") == \
+        _profile_part(target)
+    if replying and author.get("expects_reply") is False:
+        return (f"@{_profile_part(target)} sent you a reply that needs no answer, so this exchange is over. "
+                "Do not message them again in this turn; tell your user what you learned instead."), {}
+    cap = _settings().max_hops
+    ask = (_profile_part(target), hashlib.sha256(" ".join(str(body).split()).encode()).hexdigest())
+    with _LOCK:
+        state = _CHAINS.setdefault(_key(agent), _new_state(time.monotonic()))
+        hops = state["hops"] + 1
+        if hops > cap:
+            return (f"This bot-to-bot exchange already reached its limit of {cap} messages "
+                    "(gateway.bot_loop_guard.max_hops), so nothing was sent. Do not retry; finish your turn "
+                    "and tell your user where it stands."), {}
+        # Inside a bot exchange the same ask twice is one hop, not a new message (a person's turn may repeat).
+        if state["hops"] and ask in state["asks"]:
+            return (f"You already sent this exact message to @{ask[0]} in this exchange, so it was not sent "
+                    "again. Wait for their answer or say something new."), {}
+        state["asks"].add(ask)
+        state.update(sent=max(state["sent"], hops), updated=time.monotonic())
+    # Hop 1 is what a receiver assumes for an envelope without the field, so it is left out.
+    return None, {**({"hops": hops} if hops > 1 else {}), **({"expects_reply": False} if replying else {})}

agent/turn_author.py

@@ -15,6 +15,7 @@ from typing import Any, Dict, Mapping, MutableMapping, Optional
 TURN_AUTHOR_ENV = "HERMES_TURN_AUTHOR"
 
 _MAX_FIELD_LEN = 200
+_MAX_CHAIN_HOPS = 1000
 
 
 _DROPPED_CATEGORIES = frozenset({"Cc", "Cs", "Cn", "Co"})
@@ -73,6 +74,12 @@ def parse_turn_author(raw: Any) -> Optional[dict[str, Any]]:
         }
         if author["id"] is None and author["name"] is None:
             return None
+        # Runtime-written chain fields of a bot-to-bot delivery (agent/bot_chain.py); never model input.
+        hops = raw.get("hops")
+        if isinstance(hops, int) and not isinstance(hops, bool) and 0 < hops <= _MAX_CHAIN_HOPS:
+            author["hops"] = hops
+        if raw.get("expects_reply") is False:
+            author["expects_reply"] = False
         origin = _clean_text(raw.get("origin"))
         if origin and author["id"] and author["id"].startswith("bot:") and "/" not in author["id"]:
             author["id"] = _clean_text(bot_author_id(author["id"][len("bot:"):], origin))

agent/turn_context.py

@@ -1115,6 +1115,8 @@ def build_turn_context(
     # Reset first: a cached gateway agent must never carry the previous turn's bot author into a human turn.
     turn_author = parse_turn_author(turn_author)
     agent._turn_author = turn_author
+    from agent.bot_chain import on_turn_start as _bot_chain_turn_start
+    _bot_chain_turn_start(agent, turn_author, persist_user_display_kind)
 
     # Recover a rotated session before binding log/turn ids or copying client history so
     # everything in this turn belongs to the canonical child.

gateway/authz_mixin.py

@@ -704,6 +704,22 @@ class GatewayAuthorizationMixin:
             )
         return allowed
 
+    def _admit_bot_chain(self, event) -> bool:
+        """The chain rules for one admitted inbound message (``BotLoopGuard.chain_admit``), read under the
+        transport profile like the budget. A human message resets its conversation's chain; a bot message
+        past ``max_hops``, a repeat of the same ask, or a bot answering the reply this bot just gave it
+        starts no turn (#137027)."""
+        source = event.source
+        is_bot = bool(getattr(source, "is_bot", False))
+        if getattr(event, "internal", False) or (not is_bot and not self._bot_loop_guard_instance().tracking_chains):
+            return True
+        allowed, state = self._under_authorization_profile(source, lambda: self._bot_loop_guard_instance().chain_admit(
+            self._bot_loop_guard_conversation(source), str(source.user_id or ""), str(event.text or ""), is_bot=is_bot))
+        if not allowed:
+            logger.info("Bot loop guard ended a bot exchange in %s chat %s (%s from %s; gateway.bot_loop_guard).",
+                        source.platform.value if source.platform else "", source.chat_id, state, source.user_id)
+        return allowed
+
     def _is_user_authorized(self, source: SessionSource, *, allow_adapter_delegation: bool = True) -> bool:
         """Whether a user may use the bot.
 

gateway/bot_loop_guard.py

@@ -3,17 +3,23 @@
 ``{PLATFORM}_ALLOW_BOTS`` only decides admission, so two Hermes profiles replying to each other never stop.
 The guard counts admitted bot messages per conversation and drops further ones for ``cooldown_seconds``
 once ``max_events`` land inside ``window_seconds``. Settings: config.yaml ``gateway.bot_loop_guard``.
+
+The chain rules (``chain_admit``) end an exchange long before that budget: a human message starts a
+new chain; each bot message after it is one hop and ``max_hops`` caps them; a bot repeating the same
+text inside one chain is one hop, not a new ask; and with ``bot_replies_expect_reply`` false (the
+default) a bot answering the reply this bot just gave it is not answered again (#137027).
 """
 
 from __future__ import annotations
 
+import hashlib
 import threading
 import time
 from collections import deque
 from dataclasses import dataclass
 from typing import Callable, Deque, Dict, Hashable, Tuple
 
-__all__ = ["BotLoopGuard", "BotLoopGuardSettings", "load_settings", "settings_from_config"]
+__all__ = ["BotLoopGuard", "BotLoopGuardSettings", "configured_bot_user", "load_settings", "settings_from_config"]
 
 _TRUTHY = frozenset({"true", "1", "yes", "on"})
 _FALSY = frozenset({"false", "0", "no", "off"})
@@ -25,6 +31,8 @@ class BotLoopGuardSettings:
     max_events: int = 20
     window_seconds: float = 300.0
     cooldown_seconds: float = 600.0
+    max_hops: int = 4
+    bot_replies_expect_reply: bool = False
 
 
 def _as_bool(raw, default: bool) -> bool:
@@ -66,9 +74,23 @@ def settings_from_config(cfg) -> BotLoopGuardSettings:
         max_events=_as_positive_int(block.get("max_events"), defaults.max_events),
         window_seconds=_as_positive(block.get("window_seconds"), defaults.window_seconds),
         cooldown_seconds=_as_positive(block.get("cooldown_seconds"), defaults.cooldown_seconds),
+        max_hops=_as_positive_int(block.get("max_hops"), defaults.max_hops),
+        bot_replies_expect_reply=_as_bool(block.get("bot_replies_expect_reply"), defaults.bot_replies_expect_reply),
     )
 
 
+def configured_bot_user(platform_config, user_id) -> bool:
+    """Whether ``user_id`` is in the adapter's ``bot_users`` list (config.yaml platform ``extra``): the
+    opt-in that marks a peer bot on a platform whose events carry no bot flag (Mattermost, Matrix,
+    Signal, IRC, ...), so this guard and the bot-peer delivery rules engage for it (#137027)."""
+    if user_id is None or user_id == "":
+        return False
+    configured = (getattr(platform_config, "extra", None) or {}).get("bot_users")
+    if isinstance(configured, str):
+        configured = configured.split(",")
+    return isinstance(configured, (list, tuple, set)) and str(user_id) in {str(v).strip() for v in configured}
+
+
 def load_settings() -> BotLoopGuardSettings:
     """Settings from the live config.yaml. Defaults when the config cannot be read."""
     try:
@@ -94,12 +116,18 @@ class BotLoopGuard:
         self._events: dict[Hashable, deque[float]] = {}
         self._cooldown_until: dict[Hashable, float] = {}
         self._last_sweep = 0.0
+        self._chains: dict[Hashable, _Chain] = {}
 
     @property
     def tracked_conversations(self) -> int:
         with self._lock:
             return len(self._events)
 
+    @property
+    def tracking_chains(self) -> bool:
+        """Whether any bot exchange is open, so a human message has a chain to reset."""
+        return bool(self._chains)
+
     def blocked(self, conversation: Hashable) -> bool:
         """True while ``conversation`` is cooling down. Reads only, so callers may ask as often as they like."""
         if not self._settings().enabled:
@@ -129,6 +157,34 @@ class BotLoopGuard:
             events.append(now)
             return True, "ok"
 
+    def chain_admit(self, conversation: Hashable, sender: str, text: str, *, is_bot: bool) -> tuple[bool, str]:
+        """Apply the chain rules to one admitted inbound message. Returns ``(allowed, state)``; state is
+        ``disabled``, ``human`` (chain reset), ``ok``, ``max_hops``, ``duplicate`` or ``reply_to_reply``."""
+        settings = self._settings()
+        if not settings.enabled:
+            return True, "disabled"
+        now = self._clock()
+        with self._lock:
+            self._sweep(now, settings)
+            if not is_bot:
+                self._chains.pop(conversation, None)
+                return True, "human"
+            chain = self._chains.get(conversation)
+            if chain is None or now - chain.updated > settings.cooldown_seconds:
+                chain = self._chains[conversation] = _Chain()
+            chain.updated = now
+            ask = (sender, hashlib.sha256(" ".join(str(text or "").split()).encode()).hexdigest())
+            if ask in chain.asks:
+                return False, "duplicate"
+            if not settings.bot_replies_expect_reply and chain.last_trigger == sender:
+                return False, "reply_to_reply"
+            if chain.hops >= settings.max_hops:
+                return False, "max_hops"
+            chain.hops += 1
+            chain.asks.add(ask)
+            chain.last_trigger = sender
+            return True, "ok"
+
     def _sweep(self, now: float, settings: BotLoopGuardSettings) -> None:
         """Drop idle conversations at most once per window so memory stays bounded."""
         if now - self._last_sweep < settings.window_seconds:
@@ -139,3 +195,16 @@ class BotLoopGuard:
             del self._events[key]
         for key in [k for k, until in self._cooldown_until.items() if until <= now]:
             del self._cooldown_until[key]
+        for key in [k for k, chain in self._chains.items() if now - chain.updated > settings.cooldown_seconds]:
+            del self._chains[key]
+
+
+class _Chain:
+    """One bot-to-bot exchange since the last human message in a conversation."""
+    __slots__ = ("asks", "hops", "last_trigger", "updated")
+
+    def __init__(self) -> None:
+        self.hops = 0
+        self.asks: set[tuple[str, str]] = set()
+        self.last_trigger: str = ""
+        self.updated = 0.0

gateway/platforms/base.py

@@ -4723,6 +4723,8 @@ class BasePlatformAdapter(ABC):
         profile is stamped on ``source.profile`` for per-profile HERMES_HOME isolation."""
         def _opt(value) -> Optional[str]:
             return str(value) if value else None
+        from gateway.bot_loop_guard import configured_bot_user
+        is_bot = is_bot or configured_bot_user(getattr(self, "config", None), user_id)
         fields = dict(
             platform=self.platform, chat_id=str(chat_id), chat_name=chat_name, chat_type=chat_type,
             user_id=None if user_id is None or user_id == "" else str(user_id),

gateway/run_busy.py

@@ -616,7 +616,11 @@ class GatewayBusySessionMixin:
         return None if getattr(event, "internal", False) else event.message_id
 
     async def _send_busy_reply(self, event: MessageEvent, adapter, content: str, *, plain_anchor: bool = False) -> None:
-        """Send a busy-path reply anchored to the event (thread metadata included)."""
+        """Send a busy-path reply anchored to the event (thread metadata included). Never to a bot:
+        an ack, redirect or drain notice is a new inbound message for its turn loop (#137027)."""
+        if getattr(event.source, "is_bot", False):
+            logger.debug("Busy-path notice suppressed for bot-authored message in %s", event.source.chat_id)
+            return
         reply_anchor = self._reply_anchor_for_event(event)
         await adapter._send_with_retry(
             chat_id=event.source.chat_id, content=content,
@@ -922,9 +926,6 @@ class GatewayBusySessionMixin:
         return message
 
     async def _send_busy_ack_reply(self, event: MessageEvent, adapter, message: str) -> None:
-        if getattr(event.source, "is_bot", False):
-            logger.debug("Busy ack suppressed for bot-authored message in %s", event.source.chat_id)
-            return
         try:
             await self._send_busy_reply(event, adapter, message)
         except Exception as e:
@@ -953,7 +954,7 @@ class GatewayBusySessionMixin:
             )
             return True  # handled (silently dropped); do not fall through
         # A steered or queued follow-up never reaches _hm_admit_event, so the budget is charged here.
-        if not self._admit_bot_message_for_source(event.source):
+        if not (self._admit_bot_message_for_source(event.source) and self._admit_bot_chain(event)):
             return True
         event._bot_loop_admitted = True
 

gateway/run_inbound.py

@@ -245,7 +245,8 @@ class GatewayInboundMixin(GatewayInboundHooksMixin, GatewayInboundTranscriptionM
                 logger.warning("Unauthorized user: %s (%s) on %s", source.user_id, source.user_name, source.platform.value)
             return None
         # The busy path charged this event on arrival; a drained follow-up must not pay twice.
-        if not getattr(event, "_bot_loop_admitted", False) and not self._admit_bot_message_for_source(source):
+        if not getattr(event, "_bot_loop_admitted", False) and not (
+                self._admit_bot_message_for_source(source) and self._admit_bot_chain(event)):
             return None
         return event, source, False
 

gateway/run_turn.py

@@ -302,7 +302,7 @@ class GatewayTurnMixin(GatewayBackgroundMixin, GatewayTurnPrepareMixin, GatewayT
             agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure = (
                 self._hmwa_classify_turn_failure(agent_result, history, session_entry)
             )
-            if agent_failed_early and not is_context_overflow_failure:
+            if agent_failed_early and not is_context_overflow_failure and not getattr(source, "is_bot", False):
                 response = self._hmwa_add_failed_turn_notice(response, self._hmwa_failed_turn_notice(agent_result))
             response, session_entry = await self._hmwa_compression_exhaustion_reset(
                 agent_result, response, session_entry, session_key, source,
@@ -909,9 +909,12 @@ class GatewayTurnMixin(GatewayBackgroundMixin, GatewayTurnPrepareMixin, GatewayT
                         if kind in {"heartbeat", "waiting", "long_running", "status"}
                         else t("gateway.progress.status_fallback_short"))
 
-        # Webhooks can't edit messages, so tool progress / log mode are off there.
+        # Webhooks can't edit messages, so tool progress / log mode are off there. A bot peer gets the
+        # final reply only: every progress line, interim message or thinking bubble is one more inbound
+        # message for its turn loop (#137027).
         is_webhook = source.platform == Platform.WEBHOOK
-        tool_progress_enabled = progress_mode not in {"off", "log"} and not is_webhook
+        bot_peer = bool(getattr(source, "is_bot", False))
+        tool_progress_enabled = progress_mode not in {"off", "log"} and not is_webhook and not bot_peer
         # Live status for text-rendering typing indicators (Slack); independent of tool_progress.
         _live_status_mode = resolve_display_setting(user_config, platform_key, "live_status", "full")
         _live_status_adapter = (
@@ -924,10 +927,10 @@ class GatewayTurnMixin(GatewayBackgroundMixin, GatewayTurnPrepareMixin, GatewayT
         interim_assistant_messages_mode = _display_surface_mode(
             "interim_assistant_messages", default=True, require_platform_override_for={Platform.MATTERMOST},
         )
-        interim_assistant_messages_enabled = not is_webhook and interim_assistant_messages_mode != "off"
+        interim_assistant_messages_enabled = not is_webhook and not bot_peer and interim_assistant_messages_mode != "off"
         _thinking_enabled = _display_surface_mode(
             "thinking_progress", default=False, require_platform_override_for={Platform.MATTERMOST},
-        ) != "off"
+        ) != "off" and not bot_peer
         # Slack-native task cards need the progress queue even with text tool_progress off.
         # Slack-native task cards (#29483): when the Slack adapter's opt-in is set, tool progress renders as
         # native plan/task cards via chat.startStream — the progress queue is needed even though Slack keeps
@@ -942,6 +945,7 @@ class GatewayTurnMixin(GatewayBackgroundMixin, GatewayTurnPrepareMixin, GatewayT
             source.platform == Platform.SLACK
             and hasattr(adapter, "native_task_cards_enabled")
             and not (_tool_progress_explicit and progress_mode == "off")
+            and not bot_peer
         ):
             try:
                 _native_slack_task_cards = bool(adapter.native_task_cards_enabled())
@@ -2206,7 +2210,7 @@ class GatewayTurnMixin(GatewayBackgroundMixin, GatewayTurnPrepareMixin, GatewayT
         # Periodic "still working" notifications so the user knows the agent hasn't died.
         _executor_task_holder: list = [None]  # bound once the executor future exists (see below)
         _notify_task = (
-            None if (scheduled_heartbeat or turn_ctx.mute_notification_reply)
+            None if (scheduled_heartbeat or turn_ctx.mute_notification_reply or getattr(source, "is_bot", False))
             else spawn(self._run_agent_notify_long_running(disp, turn_ctx, _executor_task_holder))
         )
 

gateway/run_turn_persistence.py

@@ -126,6 +126,9 @@ class GatewayTurnPersistenceMixin:
         _silence_kind = agent_result.get("queued_terminal_display_kind", persist_user_display_kind)
         _silence_reply_expected = agent_result.get("queued_terminal_reply_expected", reply_expected)
         _silence_source = agent_result.get("queued_terminal_source", source)
+        # A bot peer expects no reply: its message may end the exchange in silence (#137027).
+        if getattr(_silence_source, "is_bot", False) and _silence_reply_expected is None:
+            _silence_reply_expected = False
         if _intentional_silence and not self._silence_allowed_for_source(
             _silence_source, _silence_kind, _silence_reply_expected, agent_result,
         ):
@@ -144,6 +147,10 @@ class GatewayTurnPersistenceMixin:
         # "(empty)" = the model produced no visible content after exhausting all retries. One
         # text with the CLI explainer and the desktop (agent/turn_explainers.py) so the user
         # reads the same words on every surface.
+        if getattr(source, "is_bot", False) and (response == "(empty)" or agent_result.get("failed")):
+            # A failed or empty turn's text is runtime failure copy, never the model's answer: a bot
+            # peer would take it as a message to answer (#137027).
+            response = ""
         if response == "(empty)" and not _intentional_silence:
             from agent.turn_explainers import EMPTY_RESPONSE_EXPLANATION
 
@@ -169,7 +176,10 @@ class GatewayTurnPersistenceMixin:
         # Fix for #18765.
         if not _intentional_silence:
             response = strip_trailing_silence_marker(response)
-            response = _normalize_empty_agent_response(agent_result, response, history_len=len(history))
+            # A bot peer gets the model's own words only: the runtime's retry hints and "not
+            # processed" lines are new inbound messages for its turn loop (#137027).
+            if not getattr(source, "is_bot", False):
+                response = _normalize_empty_agent_response(agent_result, response, history_len=len(history))
             response = _sanitize_gateway_final_response(source.platform, response)
 
         # The agent thread already updated the contextvar; propagate to SessionEntry + _save() only
@@ -713,6 +723,10 @@ class GatewayTurnPersistenceMixin:
         elif status_code == 400:
             status_hint = t("gateway.errors.hint_rejected")
         from gateway.session_results import record_unexecuted_failure
+        if getattr(source, "is_bot", False):
+            # The failure is recorded for the operator; the bot peer gets no apology to answer (#137027).
+            record_unexecuted_failure(t("gateway.errors.generic_failed"))
+            return None
         if getattr(e, "reason", None) == "launch_credentials_unavailable":
             # The session's own route key is gone or changed: /retry repeats it, /new is the way out.
             from agent.turn_failure_copy import launch_credentials_copy

gateway/run_turn_prepare.py

@@ -475,7 +475,7 @@ class GatewayTurnPrepareMixin:
         turn_sidecar_notes.append(context_note)
 
         try:
-            should_notify = reset_reason == "suspended"
+            should_notify = reset_reason == "suspended" and not getattr(source, "is_bot", False)
             adapter = self._delivery_adapter_for(source) if should_notify else None
             if adapter:
                 notice = t("gateway.session.auto_reset_notice")

gateway/run_turn_runner.py

@@ -354,7 +354,7 @@ class TurnRunner(GatewayTurnProgressMixin, GatewaySessionAgentMixin):
         sync worker thread; hop onto the gateway loop. Fired-once latch lives on the cached agent."""
         from gateway.run import render_notice_line
         from gateway.warning_notifications import is_diagnostic_notice, render_notification
-        if self._ctx.mute_notification_reply or not self._status_live():
+        if self._ctx.mute_notification_reply or not self._status_live() or getattr(self._ctx.source, "is_bot", False):
             return
         # A usage-window limit line repeats the turn's own reply, which carries the sentence and the
         # switch button (a chat message cannot be cleared, so it would only pile up). The 90% heads-up

plugins/platforms/mattermost/adapter.py

@@ -129,11 +129,10 @@ class MattermostAdapter(BasePlatformAdapter):
 
         Mattermost's posted WebSocket event contains only a sender ID, not a bot flag.  Keep the
         classification opt-in: an unknown account stays a human rather than silently changing its
-        authorization or delivery behavior.
+        authorization or delivery behavior. ``bot_users`` in config.yaml is read by the base
+        ``build_source``; ``MATTERMOST_BOT_USERS`` is its env spelling.
         """
-        extra = getattr(self.config, "extra", {}) or {}
-        configured = extra.get("bot_users", _get_scoped_secret("MATTERMOST_BOT_USERS", ""))
-        return str(user_id) in _channel_id_set(configured)
+        return str(user_id) in _channel_id_set(_get_scoped_secret("MATTERMOST_BOT_USERS", ""))
 
     # --- HTTP helpers ---
 
@@ -594,7 +593,10 @@ class MattermostAdapter(BasePlatformAdapter):
         source = self.build_source(
             chat_id=channel_id, chat_type=_CHANNEL_TYPE_MAP.get(data.get("channel_type", "O"), "channel"),
             user_id=sender_id, user_name=data.get("sender_name", "").lstrip("@") or sender_id,
-            thread_id=thread_id, message_id=post_id, is_bot=self._is_configured_bot_user(sender_id))
+            thread_id=thread_id, message_id=post_id,
+            # A bot account's post carries props.from_bot (a string "true" on the wire).
+            is_bot=str((post.get("props") or {}).get("from_bot", "")).lower() == "true"
+            or self._is_configured_bot_user(sender_id))
         from gateway.platforms.base import resolve_channel_prompt
         await self.handle_message(MessageEvent(
             text=message_text, message_type=msg_type, source=source, raw_message=post, message_id=post_id,

plugins/platforms/qqbot/adapter.py

@@ -784,7 +784,8 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
         nick = str(member.get("nick", "")) or str(author.get("username", ""))
         await self._ingest(
             d, msg_id, content, d.get("attachments"), timestamp,
-            chat_id=channel_id, qq_chat_type="guild", user_id=author_id, user_name=nick or None, chat_type="group")
+            chat_id=channel_id, qq_chat_type="guild", user_id=author_id, user_name=nick or None, chat_type="group",
+            is_bot=bool(author.get("bot")))
 
     async def _handle_dm_message(self, d, msg_id, content, author, timestamp) -> None:
         guild_id = str(d.get("guild_id", ""))
@@ -797,7 +798,8 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
             return
         await self._ingest(
             d, msg_id, content, d.get("attachments"), timestamp,
-            chat_id=guild_id, qq_chat_type="dm", user_id=author_id, chat_type="dm")
+            chat_id=guild_id, qq_chat_type="dm", user_id=author_id, chat_type="dm",
+            is_bot=bool(author.get("bot")))
 
     _INBOUND_HANDLERS = {
         "C2C_MESSAGE_CREATE": "_handle_c2c_message",