Telegram: delivery-only notification sink for notifications.telegram_chat_id groups

nerve/channels/telegram.py

@@ -1551,12 +1551,43 @@ async def _extract_zip(
 
         return blocks, "\n".join(parts)
 
+    def _is_delivery_only_sink(self, chat: Any) -> bool:
+        """True when ``chat`` is an opted-in, non-private notification sink.
+
+        ``notifications.telegram_chat_id`` can be pointed at a dedicated group
+        so pushes (the notify tool, async questions/approvals, tg-notify.sh)
+        land there instead of the owner's DM. When
+        ``notifications.delivery_only_sink`` is enabled, such a group is
+        delivery-only: inbound messages must not start an agent turn.
+
+        Gated behind the opt-in flag (default off) so an existing install that
+        used a group as the sink and still chatted there is unaffected. A DM
+        sink (chat id == the user's own private chat) stays interactive
+        regardless, hence the non-private requirement. Inline-button callbacks
+        are handled separately, so questions/approvals delivered here stay
+        answerable.
+        """
+        notif = self.config.notifications
+        if not getattr(notif, "delivery_only_sink", False):
+            return False
+        notif_chat = notif.telegram_chat_id
+        return bool(notif_chat) and chat.id == notif_chat and chat.type != "private"
+
     async def _handle_message(self, update: Update, context: Any) -> None:
         """Handle incoming text and photo messages — delegate to router."""
         self._touch()
         if not self._is_authorized(update.effective_user.id):
             return
 
+        # Delivery-only notification sink: never start an agent turn on a
+        # message that arrives in the notifications group (a one-way channel).
+        if self._is_delivery_only_sink(update.effective_chat):
+            logger.info(
+                "Ignoring inbound message in delivery-only notification chat %s",
+                update.effective_chat.id,
+            )
+            return
+
         # Media group (album) — collect all parts before processing
         if update.message.media_group_id:
             await self._collect_media_group(update)

nerve/config.py

@@ -1934,6 +1934,12 @@ class NotificationsConfig:
     """Async notification delivery settings."""
     channels: list[str] = field(default_factory=lambda: ["web", "telegram"])
     telegram_chat_id: int | None = None       # Target chat; falls back to first allowed_user
+    # Opt-in, default off = backward compatible. When true AND telegram_chat_id
+    # is a non-private group/supergroup, that chat is DELIVERY-ONLY: notifications
+    # are still sent there, but inbound messages never start an agent turn. Leave
+    # false to preserve the historical behaviour where a group set as the sink
+    # still responds to messages.
+    delivery_only_sink: bool = False
     default_expiry_hours: int = 48            # Auto-expire unanswered questions
     max_redeliveries: int = 3                 # Per-row cap on snooze/re-delivery cycles
     priority_prefixes: dict[str, str] = field(default_factory=lambda: {
@@ -1952,6 +1958,7 @@ def from_dict(cls, d: dict) -> NotificationsConfig:
         return cls(
             channels=d.get("channels", ["web", "telegram"]),
             telegram_chat_id=d.get("telegram_chat_id"),
+            delivery_only_sink=d.get("delivery_only_sink", False),
             default_expiry_hours=d.get("default_expiry_hours", 48),
             max_redeliveries=d.get("max_redeliveries", 3),
             priority_prefixes=d.get("priority_prefixes", {

tests/test_telegram_notification_sink.py

@@ -0,0 +1,72 @@
+"""Tests for the delivery-only notification-sink guard in TelegramChannel.
+
+A dedicated notification group (``notifications.telegram_chat_id`` pointed at a
+group) receives pushes but must never start an agent turn on inbound messages.
+``_is_delivery_only_sink`` is the predicate the message handler consults.
+"""
+
+from types import SimpleNamespace
+
+from nerve.channels.telegram import TelegramChannel
+
+# Obviously-synthetic ids — never a real chat.
+SINK = -1000000000001
+OTHER_GROUP = -1000000000002
+DM = 424242
+
+
+def _channel(sink_chat_id, delivery_only=True):
+    """A TelegramChannel exposing only the config the guard reads.
+
+    Bypasses ``__init__`` (which builds the whole bot application); the guard
+    depends on nothing but ``config.notifications`` (``telegram_chat_id`` plus
+    the ``delivery_only_sink`` opt-in flag).
+    """
+    ch = TelegramChannel.__new__(TelegramChannel)
+    # `config` is a read-only property returning ``self._config()`` — a
+    # zero-arg callable supplied at construction. Mirror that shape.
+    ch._config = lambda: SimpleNamespace(
+        notifications=SimpleNamespace(
+            telegram_chat_id=sink_chat_id,
+            delivery_only_sink=delivery_only,
+        ),
+    )
+    return ch
+
+
+def _chat(chat_id, chat_type):
+    return SimpleNamespace(id=chat_id, type=chat_type)
+
+
+def test_group_sink_is_delivery_only():
+    ch = _channel(SINK)  # delivery_only_sink opt-in enabled
+    assert ch._is_delivery_only_sink(_chat(SINK, "group")) is True
+    assert ch._is_delivery_only_sink(_chat(SINK, "supergroup")) is True
+
+
+def test_flag_off_keeps_group_sink_interactive():
+    # Backward-compat: with the opt-in flag OFF (the default), a group set as
+    # the sink still responds — no delivery-only behaviour — even for the sink
+    # chat itself. This is the case serxa flagged: don't break installs that
+    # use a group for both notifications and interaction.
+    ch = _channel(SINK, delivery_only=False)
+    assert ch._is_delivery_only_sink(_chat(SINK, "group")) is False
+    assert ch._is_delivery_only_sink(_chat(SINK, "supergroup")) is False
+
+
+def test_other_chats_are_not_sink():
+    ch = _channel(SINK)
+    assert ch._is_delivery_only_sink(_chat(DM, "private")) is False
+    assert ch._is_delivery_only_sink(_chat(OTHER_GROUP, "group")) is False
+
+
+def test_private_chat_with_sink_id_stays_interactive():
+    # Defensive: a DM must stay interactive even if its id equals the sink id,
+    # since the guard requires a non-private chat.
+    ch = _channel(SINK)
+    assert ch._is_delivery_only_sink(_chat(SINK, "private")) is False
+
+
+def test_no_sink_configured_disables_guard():
+    ch = _channel(None)
+    assert ch._is_delivery_only_sink(_chat(SINK, "group")) is False