fix(gateway): drop outbound silence-narration messages pre-send

Hallucinated 'silence' tokens (*(silent)*, _silent_, the bare '.', '...',
'silent', no response/reply, the mute emoji) are emitted when a persona has
nothing actionable to say. In bot-to-bot channels the receiving bot mirrors
the token back, creating a tight loop that burns API tokens and can crash a
model with 'no content after all retries'. SOUL.md/prompt rules drift across
providers and have already failed in practice, so add a substrate-level guard.

_deliver_to_platform now drops a message whose finalized content is only a
silence-narration token, logs a WARNING with platform/chat_id/truncated
content, and returns {success: True, filtered: 'silence_narration',
delivered: False} instead of calling the adapter. Single chokepoint covers
every platform adapter; the regex is anchored start/end with a 64-char guard
so prose like 'Silence is golden — here is the plan...' or 'Silent install
completed' is never dropped. Local/file delivery is a separate path and is
left untouched. Opt out via gateway.filter_silence_narration: false or the
HERMES_FILTER_SILENCE_NARRATION env override (env wins when set).

Closes #34616

gateway/config.py

@@ -474,6 +474,13 @@ class GatewayConfig:
     
     # Delivery settings
     always_log_local: bool = True  # Always save cron outputs to local files
+    # Drop outbound "silence narration" messages (e.g. *(silent)*, 🔇, a bare
+    # ".") pre-send. These are model hallucinations emitted when a persona has
+    # nothing actionable to say; in bot-to-bot channels they mirror back and
+    # forth, burning tokens and crashing models. Substrate-level guard that
+    # survives SOUL.md/prompt drift across providers. Opt out with False for
+    # raw passthrough.
+    filter_silence_narration: bool = True
 
     # STT settings
     stt_enabled: bool = True  # Whether to auto-transcribe inbound voice messages
@@ -582,6 +589,7 @@ class GatewayConfig:
             "quick_commands": self.quick_commands,
             "sessions_dir": str(self.sessions_dir),
             "always_log_local": self.always_log_local,
+            "filter_silence_narration": self.filter_silence_narration,
             "stt_enabled": self.stt_enabled,
             "group_sessions_per_user": self.group_sessions_per_user,
             "thread_sessions_per_user": self.thread_sessions_per_user,
@@ -650,6 +658,9 @@ class GatewayConfig:
             quick_commands=quick_commands,
             sessions_dir=sessions_dir,
             always_log_local=_coerce_bool(data.get("always_log_local"), True),
+            filter_silence_narration=_coerce_bool(
+                data.get("filter_silence_narration"), True
+            ),
             stt_enabled=_coerce_bool(stt_enabled, True),
             group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
             thread_sessions_per_user=_coerce_bool(thread_sessions_per_user, False),
@@ -757,6 +768,11 @@ def load_gateway_config() -> GatewayConfig:
             if "always_log_local" in yaml_cfg:
                 gw_data["always_log_local"] = yaml_cfg["always_log_local"]
 
+            if "filter_silence_narration" in yaml_cfg:
+                gw_data["filter_silence_narration"] = yaml_cfg[
+                    "filter_silence_narration"
+                ]
+
             if "unauthorized_dm_behavior" in yaml_cfg:
                 gw_data["unauthorized_dm_behavior"] = _normalize_unauthorized_dm_behavior(
                     yaml_cfg.get("unauthorized_dm_behavior"),

gateway/delivery.py

@@ -9,6 +9,8 @@ Routes messages to the appropriate destination based on:
 """
 
 import logging
+import os
+import re
 from pathlib import Path
 from datetime import datetime
 from dataclasses import dataclass
@@ -21,6 +23,32 @@ logger = logging.getLogger(__name__)
 MAX_PLATFORM_OUTPUT = 4000
 TRUNCATED_VISIBLE = 3800
 
+# Matches strings that are *only* a "silence" narration with optional markdown
+# wrappers. Covers: *(silent)*, _silent_, `silent`, ~silent~, (silent), silent,
+# 🔇, a bare ".", "…", and the whitespace/marker-padded variants seen in the
+# wild. Anchored to start/end so substantive messages that merely *contain* the
+# word "silent" are never matched.
+_SILENCE_NARRATION = re.compile(
+    r'^[\s*_~`]*\(?\s*(silent|silence|no\s+response|no\s+reply)\s*\.?\)?[\s*_~`]*$'
+    r'|^[\s*_~`]*[\U0001F507\.\u2026]+[\s*_~`]*$',
+    re.IGNORECASE,
+)
+
+
+def _is_silence_narration(content: Optional[str]) -> bool:
+    """Return True when ``content`` is *only* a silence-narration token.
+
+    Length-guarded (real messages are longer) and anchored to the whole string
+    so legitimate prose like "The deployment ran silently" or "Silence is
+    golden — here is the plan..." is never flagged.
+    """
+    if not content:
+        return False
+    stripped = content.strip()
+    if not stripped or len(stripped) > 64:  # length guard
+        return False
+    return bool(_SILENCE_NARRATION.match(stripped))
+
 from .config import Platform, GatewayConfig
 from .session import SessionSource
 
@@ -261,6 +289,18 @@ class DeliveryRouter:
         path.write_text(content)
         return path
 
+    def _filter_silence_narration_enabled(self) -> bool:
+        """Whether the outbound silence-narration filter is active.
+
+        ``HERMES_FILTER_SILENCE_NARRATION`` env var overrides config when set;
+        otherwise the ``gateway.filter_silence_narration`` config flag wins
+        (default True).
+        """
+        env = os.getenv("HERMES_FILTER_SILENCE_NARRATION")
+        if env is not None:
+            return env.strip().lower() in ("1", "true", "yes", "on")
+        return bool(getattr(self.config, "filter_silence_narration", True))
+
     async def _deliver_to_platform(
         self,
         target: DeliveryTarget,
@@ -286,6 +326,27 @@ class DeliveryRouter:
                 + f"\n\n... [truncated, full output saved to {saved_path}]"
             )
         
+        # Substrate-level anti-loop guard: drop hallucinated "silence narration"
+        # (*(silent)*, 🔇, a bare ".", etc.) before it ever reaches the adapter.
+        # In bot-to-bot channels these tokens mirror back and forth until a
+        # model crashes with "no content after all retries". Behavioral prompt
+        # rules drift across providers; this single chokepoint covers every
+        # platform adapter regardless of which persona's prompt failed.
+        # Local/file delivery (_deliver_local) is a separate path and is never
+        # filtered — saved silence has no loop risk.
+        if self._filter_silence_narration_enabled() and _is_silence_narration(content):
+            logger.warning(
+                "Dropped silence-narration outbound to %s (chat=%s): %r",
+                target.platform.value,
+                target.chat_id,
+                content[:40],
+            )
+            return {
+                "success": True,
+                "filtered": "silence_narration",
+                "delivered": False,
+            }
+
         send_metadata = dict(metadata or {})
         is_named_telegram_private_topic = False
         named_telegram_private_topic_name: Optional[str] = None

tests/gateway/test_delivery_silence_filter.py

@@ -0,0 +1,202 @@
+"""Tests for the outbound silence-narration filter (anti-loop control).
+
+See the gateway delivery path: hallucinated "silence" tokens like ``*(silent)*``
+are dropped pre-send so bot-to-bot channels can't mirror them into a token-burning
+loop that crashes a model with "no content after all retries".
+"""
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform
+from gateway.delivery import (
+    DeliveryRouter,
+    DeliveryTarget,
+    _is_silence_narration,
+)
+
+
+# --- Truth table -----------------------------------------------------------
+
+POSITIVE_CASES = [
+    "*(silent)*",
+    "*Silence.*",
+    "🔇",
+    ".",
+    "…",
+    "...",
+    "(silent)",
+    "_silent_",
+    "silent",
+    " *(silent)* ",
+    "`silent`",
+    "~silent~",
+    "Silence",
+    "no response",
+    "No Reply.",
+]
+
+NEGATIVE_CASES = [
+    "Silence is golden — here is the plan...",
+    "Silent install completed",
+    "The deployment ran silently in the background",
+    "ok",
+    "👍",
+    "Here is the result:\n\n- item one\n- item two",
+    "I have nothing to add, but here is why: the build is green.",
+    "silently",  # word boundary — trailing letters mean it isn't a bare token
+    "no responses were collected from the survey",
+    # A 64+ char string that opens with a silence token must not be dropped.
+    "silent " + "x" * 70,
+    "",
+    "   ",
+]
+
+
+@pytest.mark.parametrize("content", POSITIVE_CASES)
+def test_is_silence_narration_positive(content):
+    assert _is_silence_narration(content) is True
+
+
+@pytest.mark.parametrize("content", NEGATIVE_CASES)
+def test_is_silence_narration_negative(content):
+    assert _is_silence_narration(content) is False
+
+
+def test_is_silence_narration_none_safe():
+    assert _is_silence_narration(None) is False
+
+
+def test_length_guard_rejects_long_strings():
+    # Exactly 65 chars of dots — over the 64-char guard, so not treated as narration.
+    assert _is_silence_narration("." * 65) is False
+    assert _is_silence_narration("." * 64) is True
+
+
+# --- Integration through DeliveryRouter ------------------------------------
+
+class RecordingAdapter:
+    def __init__(self):
+        self.calls = []
+
+    async def send(self, chat_id, content, metadata=None):
+        self.calls.append({"chat_id": chat_id, "content": content, "metadata": metadata})
+        return {"success": True}
+
+
+@pytest.mark.asyncio
+async def test_silence_narration_dropped_pre_send(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("HERMES_FILTER_SILENCE_NARRATION", raising=False)
+    adapter = RecordingAdapter()
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.DISCORD: adapter})
+    target = DeliveryTarget.parse("discord:99887766")
+
+    result = await router._deliver_to_platform(target, "*(silent)*", metadata=None)
+
+    assert adapter.calls == []  # adapter.send never invoked
+    assert result == {
+        "success": True,
+        "filtered": "silence_narration",
+        "delivered": False,
+    }
+
+
+@pytest.mark.asyncio
+async def test_real_message_is_delivered(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("HERMES_FILTER_SILENCE_NARRATION", raising=False)
+    adapter = RecordingAdapter()
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.DISCORD: adapter})
+    target = DeliveryTarget.parse("discord:99887766")
+
+    result = await router._deliver_to_platform(
+        target, "Silence is golden — here is the plan...", metadata=None
+    )
+
+    assert len(adapter.calls) == 1
+    assert adapter.calls[0]["content"] == "Silence is golden — here is the plan..."
+    assert result == {"success": True}
+
+
+@pytest.mark.asyncio
+async def test_config_opt_out_lets_silence_through(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("HERMES_FILTER_SILENCE_NARRATION", raising=False)
+    adapter = RecordingAdapter()
+    config = GatewayConfig(filter_silence_narration=False)
+    router = DeliveryRouter(config, adapters={Platform.DISCORD: adapter})
+    target = DeliveryTarget.parse("discord:99887766")
+
+    result = await router._deliver_to_platform(target, "*(silent)*", metadata=None)
+
+    assert len(adapter.calls) == 1
+    assert adapter.calls[0]["content"] == "*(silent)*"
+    assert result == {"success": True}
+
+
+@pytest.mark.asyncio
+async def test_env_override_disables_filter(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_FILTER_SILENCE_NARRATION", "0")
+    adapter = RecordingAdapter()
+    # Config default is True, but env override wins.
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.DISCORD: adapter})
+    target = DeliveryTarget.parse("discord:99887766")
+
+    result = await router._deliver_to_platform(target, "🔇", metadata=None)
+
+    assert len(adapter.calls) == 1
+    assert result == {"success": True}
+
+
+@pytest.mark.asyncio
+async def test_env_override_enables_filter_over_config(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_FILTER_SILENCE_NARRATION", "1")
+    adapter = RecordingAdapter()
+    # Config says off, env override forces on.
+    config = GatewayConfig(filter_silence_narration=False)
+    router = DeliveryRouter(config, adapters={Platform.DISCORD: adapter})
+    target = DeliveryTarget.parse("discord:99887766")
+
+    result = await router._deliver_to_platform(target, "*(silent)*", metadata=None)
+
+    assert adapter.calls == []
+    assert result["filtered"] == "silence_narration"
+
+
+@pytest.mark.asyncio
+async def test_local_delivery_not_filtered(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("HERMES_FILTER_SILENCE_NARRATION", raising=False)
+    router = DeliveryRouter(GatewayConfig(), adapters={})
+
+    results = await router.deliver(
+        content="*(silent)*",
+        targets=[DeliveryTarget.parse("local")],
+        job_id="silence-job",
+    )
+
+    # Local path saved the file (no loop risk) and was not filtered.
+    local_result = results["local"]
+    assert local_result["success"] is True
+    saved_path = local_result["result"]["path"]
+    assert saved_path.endswith(".md")
+
+
+# --- Config round-trip ------------------------------------------------------
+
+def test_config_flag_defaults_true():
+    assert GatewayConfig().filter_silence_narration is True
+
+
+def test_config_from_dict_parses_flag():
+    cfg = GatewayConfig.from_dict({"filter_silence_narration": False})
+    assert cfg.filter_silence_narration is False
+
+
+def test_config_to_dict_roundtrip():
+    cfg = GatewayConfig(filter_silence_narration=False)
+    assert cfg.to_dict()["filter_silence_narration"] is False
+    restored = GatewayConfig.from_dict(cfg.to_dict())
+    assert restored.filter_silence_narration is False