fix(gateway): count each bot message once in the loop guard and consume the author variable

The Telegram adapter asks the authorization check before dispatch, the ingress gate asks it
again, and the busy path asks a third time. Each call counted one loop-guard event, so a
Telegram bot tripped the budget after a third of the configured messages. The verdict now only
refuses a chat that is cooling down. The ingress gate counts an admitted bot message once.

`parse_turn_author` treats only booleans, integers and the strings true/1/yes as a bot flag,
and returns None for an author with neither id nor name. Names keep format characters and
non-breaking spaces so emoji sequences survive. The quiet one-shot pops HERMES_TURN_AUTHOR
before the turn so tool subprocesses do not inherit it. `max_events` must be a whole positive
number. Issue numbers move out of code comments.

agent/memory_provider.py

@@ -110,8 +110,7 @@ class MemoryProvider(ABC):
         turn_author: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Persist a completed turn (non-blocking). ``messages`` is the OpenAI-style list so far.
-        ``turn_author`` (``{"id", "name", "is_bot"}``) is who wrote the user side of this turn.
-        Providers may ignore it. The manager sends it only to signatures that accept it."""
+        ``turn_author`` (``{"id", "name", "is_bot"}``) is who wrote the user side; the manager sends it only to signatures that accept it."""
 
     @abstractmethod
     def get_tool_schemas(self) -> List[Dict[str, Any]]:
@@ -127,16 +126,14 @@ class MemoryProvider(ABC):
     # -- Optional hooks (override to opt in) ---------------------------------
 
     def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None:
-        """Per-turn tick. kwargs may include remaining_tokens, model, platform, tool_count,
-        author_id, author_name, author_is_bot. The author trio names who wrote THIS turn:
-        a shared session carries several participants and other agents, so a provider keying durable
-        state on identity must read it per turn. All three are None when the transport gave no author."""
+        """Per-turn tick. kwargs may include remaining_tokens, model, platform, tool_count, author_id, author_name,
+        author_is_bot. The author trio names who wrote THIS turn (None, None, False without one): a shared session
+        carries several participants, so a provider keying durable state on identity must read it per turn."""
 
     def identity_signature(self) -> Dict[str, Any]:
-        """Identity-mapping values that must bust a cached gateway agent when they change: which
-        user/agent identity the provider writes under, alias tables, session-name prefixing. Keys are
-        namespaced by the provider; values are JSON-serializable primitives or sorted lists. The gateway
-        calls this on an uninitialized instance on every inbound message, so keep it cheap and read-only."""
+        """Identity-mapping values that must bust a cached gateway agent when they change (writer identity, alias
+        tables, session-name prefixing). Provider-namespaced keys, JSON-serializable values. The gateway calls this
+        on an uninitialized instance on every inbound message, so keep it cheap and read-only."""
         return {}
 
     def on_session_end(self, messages: List[Dict[str, Any]]) -> None:

agent/turn_author.py

@@ -8,35 +8,51 @@ from __future__ import annotations
 
 import json
 import os
-from typing import Any, Dict, Mapping, Optional
+import unicodedata
+from typing import Any, Dict, Mapping, MutableMapping, Optional
 
 TURN_AUTHOR_ENV = "HERMES_TURN_AUTHOR"
 
 _MAX_FIELD_LEN = 200
 
 
+_DROPPED_CATEGORIES = frozenset({"Cc", "Cs", "Cn", "Co"})
+_TRUTHY = frozenset({"true", "1", "yes"})
+
+
 def _clean_text(value: Any) -> Optional[str]:
-    """Strip whitespace and control characters, cap the length; None when nothing is left."""
+    """Strip whitespace, control and unassigned characters, then cap the length. None when nothing is left.
+    Format characters and non-breaking spaces stay so emoji sequences and display names survive."""
     if not isinstance(value, str):
         return None
-    text = "".join(ch for ch in value if ch.isprintable()).strip()
+    text = "".join(ch for ch in value if unicodedata.category(ch) not in _DROPPED_CATEGORIES).strip()
     if not text:
         return None
     return text[:_MAX_FIELD_LEN]
 
 
+def _bot_flag(value: Any) -> bool:
+    if isinstance(value, str):
+        return value.strip().lower() in _TRUTHY
+    return isinstance(value, (bool, int)) and bool(value)
+
+
 def parse_turn_author(raw: Any) -> Optional[Dict[str, Any]]:
-    """Normalize a dict or JSON string into ``{"id", "name", "is_bot"}``; None for anything else. Never raises."""
+    """Normalize a dict or JSON string into ``{"id", "name", "is_bot"}``; None for anything else or without id and name.
+    The id is whatever the transport knows the sender by: ``bot:<profile>`` on bot-mode deliveries, the platform user id elsewhere."""
     try:
         if isinstance(raw, (str, bytes)):
             raw = json.loads(raw)
         if not isinstance(raw, Mapping):
             return None
-        return {
+        author = {
             "id": _clean_text(raw.get("id")),
             "name": _clean_text(raw.get("name")),
-            "is_bot": bool(raw.get("is_bot")),
+            "is_bot": _bot_flag(raw.get("is_bot")),
         }
+        if author["id"] is None and author["name"] is None:
+            return None
+        return author
     except Exception:
         return None
 
@@ -46,6 +62,11 @@ def turn_author_from_env(environ: Mapping[str, str] = os.environ) -> Optional[Di
     return parse_turn_author(environ.get(TURN_AUTHOR_ENV))
 
 
+def take_turn_author_from_env(environ: MutableMapping[str, str] = os.environ) -> Optional[Dict[str, Any]]:
+    """Read and remove ``HERMES_TURN_AUTHOR`` so subprocesses started during the turn do not inherit it."""
+    return parse_turn_author(environ.pop(TURN_AUTHOR_ENV, None))
+
+
 def turn_author_env(author: Dict[str, Any]) -> Dict[str, str]:
     """The environment entry a dispatcher merges into a child's env."""
     return {TURN_AUTHOR_ENV: json.dumps(author, separators=(",", ":"))}

agent/turn_context.py

@@ -870,8 +870,7 @@ def build_turn_context(
     # Guard stdio against OSError from broken pipes (systemd/headless/daemon).
     install_safe_stdio()
 
-    # Reset first: a cached gateway agent must never carry the previous turn's bot author into a
-    # human turn. The end-of-turn memory sync reads these back.
+    # 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
 

cli.py

@@ -4056,13 +4056,14 @@ def _sync_cli_session_id_from_agent(cli) -> None:
 
 def _run_quiet_single_query(cli, effective_query):
     """Quiet (-Q) one-shot turn: run, print the response (stderr for errors/session_id), then sys.exit with the automation exit code.
-    The turn's author comes from HERMES_TURN_AUTHOR, which only a bot-to-bot dispatcher sets on this subprocess."""
-    from agent.turn_author import turn_author_from_env
+    The turn's author comes from HERMES_TURN_AUTHOR. Only a bot-to-bot dispatcher sets it, and it is consumed
+    here so tool subprocesses do not inherit it."""
+    from agent.turn_author import take_turn_author_from_env
 
     try:
         result = cli.agent.run_conversation(
             user_message=effective_query, conversation_history=cli.conversation_history,
-            turn_author=turn_author_from_env(),
+            turn_author=take_turn_author_from_env(),
         )
     except KeyboardInterrupt:
         _emit_interrupted_session_end(cli, reason="keyboard_interrupt")

gateway/authz_mixin.py

@@ -1,8 +1,8 @@
 """User-authorization mixin for ``GatewayRunner``: may this user/chat talk to the agent,
-the per-adapter DM policy, and the unauthorized-DM behavior.
+the per-adapter DM policy, the unauthorized-DM behavior, and the bot loop guard.
 
-``gateway.run`` is never imported at module import time (cycle); the one method that logs
-imports its ``logger`` lazily so records keep the ``"gateway.run"`` name.
+``gateway.run`` is never imported at module import time (cycle). The unauthorized-DM method still
+logs through ``gateway.run``'s logger so its records keep that name.
 """
 
 from __future__ import annotations
@@ -475,23 +475,31 @@ class GatewayAuthorizationMixin:
             self._warned_telegram_group_users_legacy = True
         return source.chat_id in legacy_chat_ids
 
-    def _bot_loop_guard_admits(self, source: SessionSource, adapter_profile: Optional[str]) -> bool:
-        """Count one admitted bot-authored message; False while its conversation is over budget."""
+    def _bot_loop_guard_instance(self) -> BotLoopGuard:
         guard = getattr(self, "_bot_loop_guard", None)
         if guard is None:
             with _BOT_LOOP_GUARD_INIT_LOCK:
                 guard = getattr(self, "_bot_loop_guard", None)
                 if guard is None:
                     guard = self._bot_loop_guard = BotLoopGuard()
+        return guard
+
+    def _bot_loop_guard_conversation(self, source: SessionSource) -> tuple:
+        # One budget per conversation, not per sender pair: a per-pair key would hand N bots N budgets.
         platform = source.platform.value if source.platform else ""
-        # One budget per conversation, not per sender pair: inbound only ever shows the other bots,
-        # so a per-pair key would hand N bots N budgets.
-        allowed, state = guard.admit((adapter_profile or "", platform, str(source.chat_id or "")))
+        return (self._adapter_profile_for_source(source) or "", platform, str(source.chat_id or ""))
+
+    def _admit_bot_message(self, source: SessionSource) -> bool:
+        """Count one authorized bot-authored inbound message. False when it trips the budget or the chat is cooling down.
+        The inbound handler calls this once per message; ``_is_user_authorized`` only peeks because it is asked several times."""
+        if not getattr(source, "is_bot", False):
+            return True
+        allowed, state = self._bot_loop_guard_instance().admit(self._bot_loop_guard_conversation(source))
         if state == "tripped":
             logger.warning(
                 "Bot loop guard is dropping bot messages in %s chat %s: bot %s sent one message too many "
                 "for the window, cooling down (gateway.bot_loop_guard in config.yaml).",
-                platform, source.chat_id, source.user_id,
+                source.platform.value if source.platform else "", source.chat_id, source.user_id,
             )
         return allowed
 
@@ -501,15 +509,14 @@ class GatewayAuthorizationMixin:
         Order: trusted-upstream delegation, chat-scoped group allowlists, ``{PLATFORM}_ALLOW_BOTS``,
         per-platform allow-all, adapter role auth, pairing store, env/config allowlists,
         ``GATEWAY_ALLOW_ALL_USERS``, default deny. A bot-authored message that any of these admits
-        then passes the bot loop guard.
+        is still refused while its chat's loop guard is cooling down.
         """
         if not self._principal_authorized(source, allow_adapter_delegation=allow_adapter_delegation):
             return False
         if not getattr(source, "is_bot", False):
             return True
-        # The guard judges the final verdict: the chat-scoped allowlist admits a bot in an allowlisted
-        # chat before the ALLOW_BOTS block runs, so a guard inside that block would never see it.
-        return self._bot_loop_guard_admits(source, self._adapter_profile_for_source(source))
+        # The guard judges the final verdict: a chat allowlist admits a bot before the ALLOW_BOTS block runs.
+        return not self._bot_loop_guard_instance().blocked(self._bot_loop_guard_conversation(source))
 
     def _principal_authorized(self, source: SessionSource, *, allow_adapter_delegation: bool) -> bool:
         """The allowlist verdict alone, before the bot loop guard."""

gateway/bot_loop_guard.py

@@ -1,13 +1,8 @@
-"""Sliding-window budget for bot-authored inbound messages (#91481).
+"""Sliding-window budget for bot-authored inbound messages.
 
-``{PLATFORM}_ALLOW_BOTS`` only decides admission. When two Hermes profiles reply to each
-other, every reply satisfies the ``mentions`` test again, so nothing ends the exchange.
-The guard counts admitted bot-authored messages per conversation and, once a conversation
-exceeds ``max_events`` inside ``window_seconds``, drops further bot messages there for
-``cooldown_seconds``. Human traffic never enters the guard.
-
-Settings live in config.yaml under ``gateway.bot_loop_guard``:
-``enabled`` (true), ``max_events`` (20), ``window_seconds`` (300), ``cooldown_seconds`` (600).
+``{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``.
 """
 
 from __future__ import annotations
@@ -53,8 +48,13 @@ def _as_positive(raw, default: float) -> float:
     return value if value > 0 else default
 
 
+def _as_positive_int(raw, default: int) -> int:
+    value = _as_positive(raw, 0.0)
+    return int(value) if value >= 1 and value == int(value) else default
+
+
 def settings_from_config(cfg) -> BotLoopGuardSettings:
-    """Read ``gateway.bot_loop_guard`` from a loaded config dict; unusable values keep the default."""
+    """Read ``gateway.bot_loop_guard`` from a loaded config dict. Unusable values keep the default."""
     from hermes_cli.config import cfg_get
 
     block = cfg_get(cfg, "gateway", "bot_loop_guard", default=None)
@@ -63,14 +63,14 @@ def settings_from_config(cfg) -> BotLoopGuardSettings:
     defaults = BotLoopGuardSettings()
     return BotLoopGuardSettings(
         enabled=_as_bool(block.get("enabled"), defaults.enabled),
-        max_events=int(_as_positive(block.get("max_events"), defaults.max_events)),
+        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),
     )
 
 
 def load_settings() -> BotLoopGuardSettings:
-    """Settings from the live config.yaml; defaults when the config cannot be read."""
+    """Settings from the live config.yaml. Defaults when the config cannot be read."""
     try:
         from hermes_cli.config import load_config_readonly
 
@@ -81,10 +81,7 @@ def load_settings() -> BotLoopGuardSettings:
 
 class BotLoopGuard:
     """Per-conversation sliding window with a cooldown once the budget trips. Thread-safe.
-
-    Settings are re-read on every call so a config.yaml edit takes effect without a restart;
-    ``load_config_readonly`` caches on the file signature, so the read is cheap.
-    """
+    Settings are re-read on every call so a config.yaml edit takes effect without a restart."""
 
     def __init__(
         self,
@@ -103,12 +100,16 @@ class BotLoopGuard:
         with self._lock:
             return len(self._events)
 
+    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:
+            return False
+        with self._lock:
+            return self._cooldown_until.get(conversation, 0.0) > self._clock()
+
     def admit(self, conversation: Hashable) -> Tuple[bool, str]:
         """Count one admitted bot-authored message for ``conversation``.
-
-        Returns ``(allowed, state)`` with state one of ``disabled``, ``ok``, ``tripped``
-        (this message exceeded the budget and started the cooldown) or ``cooldown``.
-        """
+        Returns ``(allowed, state)``; state is ``disabled``, ``ok``, ``tripped`` (this message started the cooldown) or ``cooldown``."""
         settings = self._settings()
         if not settings.enabled:
             return True, "disabled"

gateway/platforms/api_server.py

@@ -619,8 +619,7 @@ def _session_chat_user_message(body: Dict[str, Any], *, param: str = "message")
 
 
 def _request_turn_author(body: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-    """Normalized body ``author`` or None when absent or null. Raises ValueError when it is not an object.
-    The value is a claim by an API-key holder: it only labels the turn for memory attribution."""
+    """Normalized body ``author``, None when absent or null, ValueError when not an object. It only labels memory."""
     raw = body.get("author")
     if raw is None:
         return None
@@ -630,14 +629,6 @@ def _request_turn_author(body: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     return parse_turn_author(raw)
 
 
-def _session_chat_author(body: Dict[str, Any]) -> tuple[Optional[Dict[str, Any]], Optional["web.Response"]]:
-    """``_request_turn_author`` with the session-chat 400 envelope (code ``invalid_author``)."""
-    try:
-        return _request_turn_author(body), None
-    except ValueError as exc:
-        return None, _error_response(str(exc), 400, code="invalid_author")
-
-
 _USAGE_TOKEN_KEYS = ("input_tokens", "output_tokens", "total_tokens")
 
 
@@ -3006,9 +2997,10 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
         user_message, err = _session_chat_user_message(body)
         if err is not None:
             return None, err
-        turn_author, err = _session_chat_author(body)
-        if err is not None:
-            return None, err
+        try:
+            turn_author = _request_turn_author(body)
+        except ValueError as exc:
+            return None, _error_response(str(exc), 400, code="invalid_author")
         system_prompt = body.get("system_message") or body.get("instructions")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return None, _error_response("system_message must be a string", 400, code="invalid_system_message")
@@ -3657,7 +3649,7 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
         ``session_history_delivery`` declares #98619 session-id provenance and default-denies: only audited
         producers whose client can address the id again pass "1" (see
         ``_bind_api_server_session``).
-        ``turn_author`` only labels the turn for memory attribution; it grants nothing."""
+        ``turn_author`` only labels the turn for memory attribution. It grants nothing."""
         loop = asyncio.get_running_loop()
         # ContextVars do not follow run_in_executor threads: capture here, re-enter in _run().
         request_profile = _api_request_profile.get()

gateway/run_agent_cache.py

@@ -98,13 +98,12 @@ class GatewayAgentCacheMixin:
             out[f"memory.{key}"] = value
         return out
 
-    # Uninitialized provider instances by name, kept for the process lifetime: loading one imports
-    # its plugin module, and identity_signature() runs on every inbound message.
+    # Kept for the process lifetime: loading a provider imports its plugin module, and this runs on every inbound message.
     _MEMORY_IDENTITY_PROVIDER_MEMO: dict[str, Any] = {}
 
     @classmethod
     def _memory_provider_identity_signature(cls, provider_name: Any) -> dict[str, Any]:
-        """The active memory provider's ``identity_signature()``; ``{}`` when there is no provider,
+        """The active memory provider's ``identity_signature()``. ``{}`` when there is no provider,
         it fails to load, or the hook raises."""
         if not isinstance(provider_name, str) or not provider_name.strip():
             return {}

gateway/run_inbound.py

@@ -190,8 +190,7 @@ class GatewayInboundMixin:
                 logger.debug("Ignoring message with no user_id from %s", source.platform.value)
                 return None
             logger.warning("Unauthorized user: %s (%s) on %s", source.user_id, source.user_name, source.platform.value)
-            # In DMs: offer pairing code. In groups: silently ignore.
-            # A bot cannot pair, and answering one during a loop-guard cooldown would be outbound traffic.
+            # DMs get a pairing code, groups are ignored. A bot cannot pair, and answering one mid-cooldown is outbound traffic.
             if (
                 source.chat_type == "dm"
                 and not getattr(source, "is_bot", False)
@@ -199,6 +198,8 @@ class GatewayInboundMixin:
             ):
                 await self._hm_offer_pairing_code(source)
             return None
+        if not self._admit_bot_message(source):
+            return None
         return event, source, False
 
     def _hm_estop_turn_allowed(self, event: "MessageEvent", source: SessionSource) -> bool:

hermes_cli/config_defaults.py

@@ -1916,7 +1916,7 @@ DEFAULT_CONFIG = {
         "loop_watchdog_probe_interval_s": 30.0,
         "loop_watchdog_probe_timeout_s": 10.0,
         "loop_watchdog_max_strikes": 3,
-        # Bot-to-bot loop guard: admitted bot messages per conversation before a cooldown (#91481).
+        # Bot-to-bot loop guard: admitted bot messages per conversation before a cooldown.
         "bot_loop_guard": {"enabled": True, "max_events": 20, "window_seconds": 300, "cooldown_seconds": 600},
         # Startup-liveness watchdog: stdlib-only daemon thread armed at process entry that
         # hard-exits 75 if the loop isn't live within the deadline. Armed before config loads, so

run_agent.py

@@ -404,7 +404,7 @@ class AIAgent(
 
         # Turn counter (added after reset_session_state was first written — #2635)
         self._user_turn_count = 0
-        # Who wrote the current turn; build_turn_context() sets it at the start of every turn.
+        # Who wrote the current turn. build_turn_context() sets it at the start of every turn.
         self._turn_author = None
         # Copilot x-initiator: True for the first API call of a user turn, False for tool-loop follow-ups.
         self._is_user_initiated_turn = False
@@ -893,7 +893,7 @@ class AIAgent(
             return
         try:
             sync_kwargs = {"session_id": self.session_id or "", **({"messages": messages} if messages is not None else {})}
-            # Stashed by build_turn_context() for this turn; None on a human turn.
+            # Stashed by build_turn_context() for this turn, None on a human turn.
             turn_author = getattr(self, "_turn_author", None)
             if turn_author is not None:
                 sync_kwargs["turn_author"] = turn_author

tests/agent/test_memory_provider.py

@@ -278,8 +278,8 @@ class TestMemoryManager:
 
         assert legacy_provider.synced_turns == [("user", "assistant")]
 
-    def test_sync_all_forwards_author_and_scope_only_to_providers_that_accept_them(self):
-        """A bot turn reaches the new-signature provider with its author; legacy and messages-only
+    def test_sync_all_forwards_author_only_to_providers_that_accept_it(self):
+        """The author reaches the new-signature provider (None on a human turn). Legacy and messages-only
         providers get the call without the keywords they cannot take."""
         legacy = FakeMemoryProvider("legacy")
         messages_only = MessagesMemoryProvider("messages")
@@ -291,27 +291,17 @@ class TestMemoryManager:
             mgr = MemoryManager()
             mgr.add_provider(p)
             mgr.sync_all("user", "assistant", session_id="s1", turn_author=author)
+            mgr.sync_all("user", "assistant")
             mgr.flush_pending(timeout=5)
 
-        assert legacy.synced_turns == [("user", "assistant")]
-        assert messages_only.synced_turns == [("user", "assistant", "s1", None)]
-        assert author_aware.synced_turns == [("user", "assistant", author)]
-
-    def test_sync_all_without_author_sends_none_to_author_aware_provider(self):
-        mgr = MemoryManager()
-        author_aware = AuthorMemoryProvider("author")
-        mgr.add_provider(author_aware)
-
-        mgr.sync_all("user", "assistant")
-        mgr.flush_pending(timeout=5)
-
-        assert author_aware.synced_turns == [("user", "assistant", None)]
+        assert legacy.synced_turns == [("user", "assistant")] * 2
+        assert messages_only.synced_turns == [("user", "assistant", "s1", None), ("user", "assistant", "", None)]
+        assert author_aware.synced_turns == [("user", "assistant", author), ("user", "assistant", None)]
 
     def test_provider_sync_accepts_inspects_named_keyword(self):
         assert MemoryManager._provider_sync_accepts(AuthorMemoryProvider(), "turn_author")
         assert not MemoryManager._provider_sync_accepts(MessagesMemoryProvider(), "turn_author")
         assert not MemoryManager._provider_sync_accepts(FakeMemoryProvider(), "messages")
-        assert MemoryManager._provider_sync_accepts_messages(MessagesMemoryProvider())
 
     # -- Tool routing -------------------------------------------------------
 

tests/agent/test_turn_author.py

@@ -8,44 +8,42 @@ from agent.turn_author import (
     TURN_AUTHOR_ENV,
     a2a_key,
     parse_turn_author,
+    take_turn_author_from_env,
     turn_author_env,
     turn_author_from_env,
 )
 
+FAMILY = "\U0001F468\u200D\U0001F469\u200D\U0001F467"
+
 
 class TestParseTurnAuthor:
-    def test_dict_is_normalized(self):
-        out = parse_turn_author({"id": "  bot:alpha ", "name": "Alpha", "is_bot": True, "extra": 1})
-        assert out == {"id": "bot:alpha", "name": "Alpha", "is_bot": True}
+    @pytest.mark.parametrize("raw, expected", [
+        ({"id": "  bot:alpha ", "name": "Alpha", "is_bot": True, "extra": 1}, {"id": "bot:alpha", "name": "Alpha", "is_bot": True}),
+        (json.dumps({"id": "bot:alpha", "name": "Alpha", "is_bot": 1}), {"id": "bot:alpha", "name": "Alpha", "is_bot": True}),
+        ({"name": "Alpha"}, {"id": None, "name": "Alpha", "is_bot": False}),
+        ({"id": 7, "name": "Alpha", "is_bot": "yes"}, {"id": None, "name": "Alpha", "is_bot": True}),
+        ({"id": "", "name": " Alpha "}, {"id": None, "name": "Alpha", "is_bot": False}),
+        ({"id": "bot:\x00al\x1bpha\n", "name": "Al\tpha\r"}, {"id": "bot:alpha", "name": "Alpha", "is_bot": False}),
+        ({"id": "bot:alpha", "name": f"{FAMILY} Al\u00a0pha\u00a0"}, {"id": "bot:alpha", "name": f"{FAMILY} Al\u00a0pha", "is_bot": False}),
+        ({"id": "x" * 500, "name": "y" * 201}, {"id": "x" * 200, "name": "y" * 200, "is_bot": False}),
+    ], ids=["dict", "json string", "missing id", "non-string id", "empty id", "control characters stripped",
+            "format characters and nbsp survive", "oversize fields capped"])
+    def test_fields_are_normalized(self, raw, expected):
+        assert parse_turn_author(raw) == expected
 
-    def test_json_string_is_parsed(self):
-        raw = json.dumps({"id": "bot:alpha", "name": "Alpha", "is_bot": 1})
-        assert parse_turn_author(raw) == {"id": "bot:alpha", "name": "Alpha", "is_bot": True}
-
-    def test_missing_fields_default(self):
-        assert parse_turn_author({}) == {"id": None, "name": None, "is_bot": False}
-
-    @pytest.mark.parametrize("raw", [None, 42, [], ["bot:alpha"], "not json", '"a string"', "[1, 2]", b"\xff"])
-    def test_junk_returns_none(self, raw):
+    @pytest.mark.parametrize("raw", [
+        None, 42, [], ["bot:alpha"], "not json", '"a string"', "[1, 2]", b"\xff",
+        {}, {"is_bot": True}, {"id": "", "name": "   "},
+    ])
+    def test_junk_and_authors_without_id_or_name_return_none(self, raw):
         assert parse_turn_author(raw) is None
 
-    def test_non_string_fields_become_none(self):
-        assert parse_turn_author({"id": 7, "name": ["x"], "is_bot": "yes"}) == {
-            "id": None, "name": None, "is_bot": True,
-        }
-
-    def test_empty_and_whitespace_become_none(self):
-        assert parse_turn_author({"id": "", "name": "   "}) == {"id": None, "name": None, "is_bot": False}
-
-    def test_control_characters_are_stripped(self):
-        out = parse_turn_author({"id": "bot:\x00al\x1bpha\n", "name": "Al\tpha\r"})
-        assert out["id"] == "bot:alpha"
-        assert out["name"] == "Alpha"
-
-    def test_oversize_fields_are_capped(self):
-        out = parse_turn_author({"id": "x" * 500, "name": "y" * 201})
-        assert len(out["id"]) == 200
-        assert len(out["name"]) == 200
+    @pytest.mark.parametrize("flag, expected", [
+        *((flag, True) for flag in (True, 1, "true", "1", " YES ")),
+        *((flag, False) for flag in (False, 0, None, "false", "0", "no", "", "bot", [True], {"a": 1}, 1.0)),
+    ])
+    def test_bot_flag_accepts_only_booleans_and_truthy_strings(self, flag, expected):
+        assert parse_turn_author({"id": "bot:alpha", "is_bot": flag})["is_bot"] is expected
 
 
 class TestEnvCarrier:
@@ -58,11 +56,16 @@ class TestEnvCarrier:
     def test_env_json_is_compact(self):
         assert turn_author_env({"id": "a", "is_bot": True})[TURN_AUTHOR_ENV] == '{"id":"a","is_bot":true}'
 
-    def test_absent_env_is_none(self):
-        assert turn_author_from_env({}) is None
+    @pytest.mark.parametrize("env", [{}, {TURN_AUTHOR_ENV: "{not json"}])
+    def test_absent_or_garbage_env_is_none(self, env):
+        assert turn_author_from_env(env) is None
 
-    def test_garbage_env_is_none(self):
-        assert turn_author_from_env({TURN_AUTHOR_ENV: "{not json"}) is None
+    def test_take_removes_the_variable(self):
+        author = {"id": "bot:alpha", "name": "Alpha", "is_bot": True}
+        env = dict(turn_author_env(author), OTHER="kept")
+        assert take_turn_author_from_env(env) == author
+        assert env == {"OTHER": "kept"}
+        assert take_turn_author_from_env(env) is None
 
 
 @pytest.mark.parametrize("author, expected", [

tests/cli/test_quiet_turn_author.py

@@ -1,12 +1,13 @@
 """``hermes chat -Q`` passes the dispatcher's HERMES_TURN_AUTHOR to ``run_conversation`` as ``turn_author``.
 
-A bot-to-bot delivery runs the recipient's turn as a ``-Q`` subprocess with that variable set;
-a human's ``-Q`` run has it unset and the turn stays unattributed.
+A bot-to-bot delivery runs the recipient's turn as a ``-Q`` subprocess with that variable set.
+A human's ``-Q`` run has it unset and the turn stays unattributed.
 """
 
 from __future__ import annotations
 
 import json
+import os
 from types import SimpleNamespace
 
 import pytest
@@ -51,6 +52,24 @@ def test_quiet_one_shot_passes_turn_author_from_env(monkeypatch, capsys):
     assert capsys.readouterr().out.strip() == "ok"
 
 
+def test_quiet_one_shot_consumes_the_variable_before_the_turn(monkeypatch):
+    """Tool subprocesses spawned during the turn must not see the dispatcher's author."""
+    author = {"id": "bot:coder", "name": "coder", "is_bot": True}
+    seen = {}
+
+    def run_conversation(**kwargs):
+        seen["env"] = os.environ.get(TURN_AUTHOR_ENV)
+        return {"final_response": "ok"}
+
+    monkeypatch.setenv(TURN_AUTHOR_ENV, json.dumps(author))
+    fake = _fake_cli([])
+    fake.agent.run_conversation = run_conversation
+    with pytest.raises(SystemExit):
+        cli._run_quiet_single_query(fake, "hello")
+    assert seen["env"] is None
+    assert TURN_AUTHOR_ENV not in os.environ
+
+
 def test_quiet_one_shot_without_env_passes_none(monkeypatch):
     assert _run(monkeypatch, None)["turn_author"] is None
 

tests/gateway/test_bot_loop_guard.py

@@ -1,13 +1,11 @@
-"""Bot-to-bot loop guard in ``_is_user_authorized`` (#91481).
+"""Bot-to-bot loop guard: ``_is_user_authorized`` refuses a chat in cooldown, ``_admit_bot_message`` counts.
 
-The scenarios run with ``TELEGRAM_GROUP_ALLOWED_CHATS`` set: that allowlist admits every sender in
-the chat, bots included, before the ``ALLOW_BOTS`` block runs. A guard hooked only into the
-``ALLOW_BOTS`` block never sees a bot in an allowlisted group, which is the configuration that
-produced the incident.
+The scenarios set ``TELEGRAM_GROUP_ALLOWED_CHATS``: that allowlist admits a bot before the ``ALLOW_BOTS``
+block runs, which is the configuration that produced the incident.
 """
 
 import logging
-import threading
+from concurrent.futures import ThreadPoolExecutor
 from types import SimpleNamespace
 
 import pytest
@@ -19,20 +17,14 @@ GROUP_CHAT = "-1001234567890"
 OTHER_GROUP = "-1009876543210"
 BOT_A = "111111111"
 BOT_B = "222222222"
+BOT_C = "333333333"
 HUMAN = "100200300"
 
 
 @pytest.fixture(autouse=True)
 def _isolate_telegram_env(monkeypatch):
-    for var in (
-        "TELEGRAM_ALLOW_BOTS",
-        "TELEGRAM_ALLOWED_USERS",
-        "TELEGRAM_ALLOW_ALL_USERS",
-        "TELEGRAM_GROUP_ALLOWED_USERS",
-        "TELEGRAM_GROUP_ALLOWED_CHATS",
-        "GATEWAY_ALLOW_ALL_USERS",
-        "GATEWAY_ALLOWED_USERS",
-    ):
+    for var in ("TELEGRAM_ALLOW_BOTS", "TELEGRAM_ALLOWED_USERS", "TELEGRAM_ALLOW_ALL_USERS", "TELEGRAM_GROUP_ALLOWED_USERS",
+                "TELEGRAM_GROUP_ALLOWED_CHATS", "GATEWAY_ALLOW_ALL_USERS", "GATEWAY_ALLOWED_USERS"):
         monkeypatch.delenv(var, raising=False)
 
 
@@ -59,17 +51,13 @@ def runner(clock, settings):
 
 
 def _bot(user_id: str, chat_id: str = GROUP_CHAT, chat_type: str = "group") -> SessionSource:
-    return SessionSource(
-        platform=Platform.TELEGRAM, chat_id=chat_id, chat_type=chat_type,
-        user_id=user_id, user_name=f"Bot{user_id}", is_bot=True,
-    )
+    return SessionSource(platform=Platform.TELEGRAM, chat_id=chat_id, chat_type=chat_type, user_id=user_id,
+                         user_name=f"Bot{user_id}", is_bot=True)
 
 
-def _human(user_id: str = HUMAN, chat_id: str = GROUP_CHAT, chat_type: str = "group") -> SessionSource:
-    return SessionSource(
-        platform=Platform.TELEGRAM, chat_id=chat_id, chat_type=chat_type,
-        user_id=user_id, user_name="Alice", is_bot=False,
-    )
+def _human(chat_id: str = GROUP_CHAT, chat_type: str = "group") -> SessionSource:
+    return SessionSource(platform=Platform.TELEGRAM, chat_id=chat_id, chat_type=chat_type, user_id=HUMAN,
+                         user_name="Alice", is_bot=False)
 
 
 def _incident_config(monkeypatch):
@@ -79,60 +67,88 @@ def _incident_config(monkeypatch):
     monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_CHATS", f"{GROUP_CHAT},{OTHER_GROUP}")
 
 
+def _inbound(runner, source: SessionSource) -> bool:
+    """What ``_hm_admit_event`` does per message: the verdict, then one count for an admitted bot."""
+    return runner._is_user_authorized(source) and runner._admit_bot_message(source)
+
+
 def _ping_pong(runner, turns: int, chat_id: str = GROUP_CHAT) -> list:
-    return [runner._is_user_authorized(_bot(BOT_A if t % 2 == 0 else BOT_B, chat_id)) for t in range(turns)]
+    return [_inbound(runner, _bot(BOT_A if t % 2 == 0 else BOT_B, chat_id)) for t in range(turns)]
 
 
 # --- authz integration -----------------------------------------------------
 
 
-def test_ping_pong_of_40_turns_is_cut_at_the_budget(monkeypatch, runner):
+def test_one_inbound_is_counted_once_however_often_the_verdict_is_asked(monkeypatch, runner):
+    """The Telegram adapter, the ingress gate and the busy path all ask the verdict for one message."""
     _incident_config(monkeypatch)
+    for _ in range(20):
+        bot = _bot(BOT_A)
+        assert [runner._is_user_authorized(bot) for _ in range(3)] == [True, True, True]
+        assert runner._admit_bot_message(bot) is True
+    assert runner._is_user_authorized(_bot(BOT_B)) is True
+    assert runner._admit_bot_message(_bot(BOT_B)) is False
+    assert runner._is_user_authorized(_bot(BOT_B)) is False
 
-    verdicts = _ping_pong(runner, 40)
+
+@pytest.mark.asyncio
+async def test_ingress_gate_counts_an_authorized_bot_once_and_drops_it_when_refused():
+    from gateway.platforms.base import MessageEvent
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner._scale_to_zero_note_real_inbound = lambda: None
+    runner._hm_pre_gateway_dispatch_hook = lambda event, source: event
+    runner._is_user_authorized_for_source = lambda source, **kw: True
+    admitted = []
+    runner._admit_bot_message = lambda source: admitted.append(source.user_id) or source.user_id != BOT_B
+
+    event = MessageEvent(text="hi", message_id="m1", source=_bot(BOT_A))
+    assert (await runner._hm_admit_event(event))[0] is event
+    assert admitted == [BOT_A]
+    assert await runner._hm_admit_event(MessageEvent(text="hi", message_id="m2", source=_bot(BOT_B))) is None
+    assert admitted == [BOT_A, BOT_B]
+
+
+@pytest.mark.parametrize("senders, chat_id, chat_type, group_allowlist", [
+    ([BOT_A, BOT_B], GROUP_CHAT, "group", True),
+    ([BOT_A, BOT_B, BOT_C], GROUP_CHAT, "group", True),
+    ([BOT_A], "123", "dm", False),
+], ids=["ping-pong of 40 turns", "three bots share one budget", "plain ALLOW_BOTS dm without a chat allowlist"])
+def test_admitted_bot_traffic_is_cut_at_the_budget(monkeypatch, runner, senders, chat_id, chat_type, group_allowlist):
+    _incident_config(monkeypatch)
+    if not group_allowlist:
+        monkeypatch.delenv("TELEGRAM_GROUP_ALLOWED_CHATS")
+
+    verdicts = [_inbound(runner, _bot(senders[i % len(senders)], chat_id, chat_type)) for i in range(40)]
 
     assert verdicts[:20] == [True] * 20
-    assert verdicts[20] is False
     assert not any(verdicts[20:])
 
 
-def test_human_in_same_group_stays_authorized_during_cooldown(monkeypatch, runner, clock):
+def test_humans_are_never_metered_and_stay_authorized_during_cooldown(monkeypatch, runner, clock):
     _incident_config(monkeypatch)
-    _ping_pong(runner, 25)
+    for _ in range(50):
+        assert _inbound(runner, _human(chat_id="123", chat_type="dm")) is True
+    assert runner._bot_loop_guard.tracked_conversations == 0
 
+    _ping_pong(runner, 25)
     assert runner._is_user_authorized(_bot(BOT_A)) is False
     assert runner._is_user_authorized(_human()) is True
     clock.advance(5)
     assert runner._is_user_authorized(_human()) is True
 
 
-def test_human_traffic_is_never_metered(monkeypatch, runner):
-    monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "mentions")
-    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", HUMAN)
-
-    for _ in range(50):
-        assert runner._is_user_authorized(_human(chat_id="123", chat_type="dm")) is True
-
-    assert runner._bot_loop_guard.tracked_conversations == 0
-
-
-def test_three_bots_in_one_group_share_one_budget(monkeypatch, runner):
-    _incident_config(monkeypatch)
-    senders = [BOT_A, BOT_B, "333333333"]
-
-    verdicts = [runner._is_user_authorized(_bot(senders[i % 3])) for i in range(30)]
-
-    assert verdicts[:20] == [True] * 20
-    assert not any(verdicts[20:])
-
-
-def test_other_group_has_its_own_budget(monkeypatch, runner):
+def test_budget_is_scoped_by_chat_and_platform(monkeypatch, runner):
     _incident_config(monkeypatch)
     _ping_pong(runner, 25)
     assert runner._is_user_authorized(_bot(BOT_A)) is False
 
-    assert runner._is_user_authorized(_bot(BOT_A, chat_id=OTHER_GROUP)) is True
-    assert runner._is_user_authorized(_bot(BOT_B, chat_id=OTHER_GROUP)) is True
+    assert _inbound(runner, _bot(BOT_A, chat_id=OTHER_GROUP)) is True
+    assert _inbound(runner, _bot(BOT_B, chat_id=OTHER_GROUP)) is True
+    monkeypatch.setenv("DISCORD_ALLOW_BOTS", "all")
+    discord_bot = SessionSource(platform=Platform.DISCORD, chat_id=GROUP_CHAT, chat_type="group", user_id=BOT_A, is_bot=True)
+    assert _inbound(runner, discord_bot) is True
 
 
 def test_slow_traffic_never_trips(monkeypatch, runner, clock):
@@ -140,7 +156,7 @@ def test_slow_traffic_never_trips(monkeypatch, runner, clock):
 
     verdicts = []
     for turn in range(60):
-        verdicts.append(runner._is_user_authorized(_bot(BOT_A if turn % 2 == 0 else BOT_B)))
+        verdicts.append(_inbound(runner, _bot(BOT_A if turn % 2 == 0 else BOT_B)))
         clock.advance(10)
 
     assert all(verdicts)
@@ -151,7 +167,7 @@ def test_window_expiry_readmits_without_tripping(monkeypatch, runner, clock):
     assert all(_ping_pong(runner, 20))
 
     clock.advance(61)
-    assert runner._is_user_authorized(_bot(BOT_A)) is True
+    assert _inbound(runner, _bot(BOT_A)) is True
 
 
 def test_cooldown_expiry_readmits_with_a_fresh_budget(monkeypatch, runner, clock):
@@ -161,7 +177,7 @@ def test_cooldown_expiry_readmits_with_a_fresh_budget(monkeypatch, runner, clock
 
     clock.advance(61)
     assert all(_ping_pong(runner, 20))
-    assert runner._is_user_authorized(_bot(BOT_A)) is False
+    assert _inbound(runner, _bot(BOT_A)) is False
 
 
 def test_disabled_via_config_admits_everything(monkeypatch, runner, settings):
@@ -171,39 +187,16 @@ def test_disabled_via_config_admits_everything(monkeypatch, runner, settings):
     assert all(_ping_pong(runner, 40))
 
 
-def test_bot_admitted_by_allow_bots_in_a_dm_is_metered(monkeypatch, runner):
-    """The plain ALLOW_BOTS path (no chat allowlist) is covered too."""
-    monkeypatch.setenv("TELEGRAM_ALLOW_BOTS", "mentions")
-    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", HUMAN)
-
-    verdicts = [runner._is_user_authorized(_bot(BOT_A, chat_id="123", chat_type="dm")) for _ in range(25)]
-
-    assert verdicts[:20] == [True] * 20
-    assert not any(verdicts[20:])
-
-
 def test_rejected_bot_messages_do_not_consume_budget(monkeypatch, runner):
     """Only admitted bot traffic counts, so an unauthorized bot cannot silence an authorized one."""
     monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", HUMAN)
 
     for _ in range(30):
-        assert runner._is_user_authorized(_bot(BOT_A, chat_id="123", chat_type="dm")) is False
+        assert _inbound(runner, _bot(BOT_A, chat_id="123", chat_type="dm")) is False
 
     assert runner._bot_loop_guard.tracked_conversations == 0
 
 
-def test_platform_scopes_the_budget_key(monkeypatch, runner):
-    _incident_config(monkeypatch)
-    _ping_pong(runner, 21)
-    assert runner._is_user_authorized(_bot(BOT_A)) is False
-
-    monkeypatch.setenv("DISCORD_ALLOW_BOTS", "all")
-    discord_bot = SessionSource(
-        platform=Platform.DISCORD, chat_id=GROUP_CHAT, chat_type="group", user_id=BOT_A, is_bot=True,
-    )
-    assert runner._is_user_authorized(discord_bot) is True
-
-
 def test_tripping_logs_one_operator_warning(monkeypatch, runner, caplog):
     _incident_config(monkeypatch)
 
@@ -222,7 +215,7 @@ def test_guard_is_created_lazily_on_a_bare_runner(monkeypatch):
     runner = object.__new__(GatewayRunner)
     runner.pairing_store = SimpleNamespace(is_approved=lambda *_a, **_kw: False)
 
-    assert runner._is_user_authorized(_bot(BOT_A, chat_id="123", chat_type="dm")) is True
+    assert _inbound(runner, _bot(BOT_A, chat_id="123", chat_type="dm")) is True
     assert isinstance(runner._bot_loop_guard, BotLoopGuard)
 
 
@@ -242,20 +235,9 @@ def test_admit_states(clock):
 
 def test_concurrent_admits_respect_the_budget(clock):
     guard = BotLoopGuard(settings=lambda: BotLoopGuardSettings(max_events=20, window_seconds=60, cooldown_seconds=60), clock=clock.now)
-    allowed = []
-    lock = threading.Lock()
 
-    def worker():
-        for _ in range(10):
-            ok, _state = guard.admit("c")
-            with lock:
-                allowed.append(ok)
-
-    threads = [threading.Thread(target=worker) for _ in range(8)]
-    for t in threads:
-        t.start()
-    for t in threads:
-        t.join()
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        allowed = list(pool.map(lambda _: guard.admit("c")[0], range(80)))
 
     assert sum(allowed) == 20
 
@@ -283,6 +265,12 @@ def test_settings_defaults_when_block_missing_or_malformed():
     assert settings_from_config({"gateway": {"bot_loop_guard": {"enabled": "maybe", "max_events": True}}}) == defaults
 
 
+@pytest.mark.parametrize("raw, expected", [("7", 7), (7.0, 7), (0.5, 20), (0, 20), (-3, 20), ("many", 20), (True, 20)])
+def test_settings_max_events_is_a_whole_positive_number(raw, expected):
+    cfg = {"gateway": {"bot_loop_guard": {"max_events": raw}}}
+    assert settings_from_config(cfg).max_events == expected
+
+
 def test_settings_parse_configured_values():
     parsed = settings_from_config({"gateway": {"bot_loop_guard": {
         "enabled": "false", "max_events": "7", "window_seconds": 30, "cooldown_seconds": 45.5,

tests/tools/test_bot_mode_dm.py

@@ -212,7 +212,7 @@ def _capture_spawn(monkeypatch):
 
 
 def _runner_parts(command):
-    """(mode, dm_file, transport argv) of a runner command; the optional ``--author <json>`` pair is skipped."""
+    """(mode, dm_file, transport argv) of a runner command. The optional ``--author <json>`` pair is skipped."""
     parts = shlex.split(command)
     marker = parts.index("--run-delivery")
     if parts[marker + 1] == "--author":

tui_gateway/methods_bot_relay.py

@@ -102,8 +102,7 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
         def _detail(p) -> str:
             return (p.stderr or p.stdout or "").strip()[-500:]
 
-        # The Desktop forwards the envelope's sender fields; the turn's author labels memory only
-        # and grants nothing. Absent fields leave the turn unattributed, as before.
+        # The Desktop forwards the envelope's sender. The author labels memory only and grants nothing.
         from tools.bot_relay import delivery_env, delivery_turn_author
         turn_env = delivery_env(delivery_turn_author(params.get("from_profile"), params.get("from_handle")))