Add per-chat passive response policy controls

Keep passive memory extraction active when response policy suppresses replies.

Co-Authored-By: GPT-5 Codex <noreply@openai.com>

Author: dcramer

docs/src/content/docs/configuration/reference.mdx

@@ -323,6 +323,8 @@ group_mode = "mention"
 
 [telegram.passive]
 enabled = false
+response_allowed_chats = []
+response_blocked_chats = []
 
 [brave_search]
 api_key = "..."

docs/src/content/docs/systems/providers.mdx

@@ -26,6 +26,8 @@ group_mode = "mention"                 # mention | always
 
 [telegram.passive]
 enabled = false                         # Passive group listening
+response_allowed_chats = []            # Optional passive-response allowlist
+response_blocked_chats = []            # Suppress passive responses in specific chats
 ```
 
 Start runtime:
@@ -88,6 +90,8 @@ Enable it explicitly:
 ```toml
 [telegram.passive]
 enabled = true
+# response_allowed_chats = ["-100123456789"]  # Reply only in these groups
+# response_blocked_chats = ["-100999999999"]  # Never reply in these groups
 ```
 
 ## Reference (Advanced)

specs/telegram.md

@@ -144,6 +144,11 @@ webhook_path = "/telegram/webhook"
 # Group chat settings
 allowed_groups = []  # Group IDs (empty = allow all with authorized users)
 group_mode = "mention"  # "mention" (default) or "always"
+
+[telegram.passive]
+enabled = false
+response_allowed_chats = []  # Optional allowlist for passive responses
+response_blocked_chats = []  # Always suppress passive responses in these chats
 ```
 
 ## Message UX
@@ -234,6 +239,7 @@ attribution in the user-visible response.
 | Group message (always mode) | Respond to all messages from authorized users |
 | Group message with mention | Strip @botname from text before processing |
 | Group not in allowed_groups | Ignore message silently |
+| Passive message in blocked chat | Skip passive engagement but still run passive memory extraction |
 
 ## Errors
 

src/ash/config/models.py

@@ -79,6 +79,13 @@ class PassiveListeningConfig(BaseModel):
     memory_lookup_timeout: float = 2.0  # seconds
     memory_similarity_threshold: float = 0.4
 
+    # Chat-level passive response policy (Telegram group chat IDs as strings)
+    # Empty allowed list means all chats are eligible for passive responses.
+    # Blocked chats always suppress passive responses.
+    # Passive memory extraction still runs for passively observed messages.
+    response_allowed_chats: list[str] = []
+    response_blocked_chats: list[str] = []
+
 
 class TelegramConfig(BaseModel):
     """Configuration for Telegram provider."""

src/ash/providers/telegram/handlers/passive_handler.py

@@ -137,6 +137,20 @@ def is_enabled(self) -> bool:
         """Check if passive listening is enabled and initialized."""
         return self._passive_decider is not None and self._memory_manager is not None
 
+    def _is_passive_response_allowed_for_chat(self, chat_id: str) -> bool:
+        """Apply chat-level policy for passive response decisions."""
+        passive_config = self._provider.passive_config
+        if not passive_config:
+            return True
+
+        if chat_id in passive_config.response_blocked_chats:
+            return False
+
+        if passive_config.response_allowed_chats:
+            return chat_id in passive_config.response_allowed_chats
+
+        return True
+
     async def handle_passive_message(self, message: IncomingMessage) -> None:
         """Handle a passively observed message (not mentioned or replied to).
 
@@ -165,6 +179,23 @@ async def handle_passive_message(self, message: IncomingMessage) -> None:
             chat_title or chat_id[:8],
         )
 
+        if not self._is_passive_response_allowed_for_chat(chat_id):
+            if self._passive_extractor:
+                asyncio.create_task(
+                    self._extract_passive_memories(message),
+                    name=f"passive_extract_{message.id}",
+                )
+
+            logger.info(
+                "passive_engagement_skipped",
+                extra={
+                    "decision_path": "response_policy",
+                    "engagement_reason": "response_policy",
+                    "username": message.username or message.user_id,
+                },
+            )
+            return
+
         # Build bot context for identity awareness (needed for name check)
         bot_context = BotContext(
             name=self._get_bot_display_name(),

tests/test_providers.py

@@ -790,6 +790,62 @@ async def mock_stream():
         assert getattr(engaging, "decision_path", None) == "name_mentioned_fast_path"
         assert getattr(engaging, "engagement_reason", None) == "name_mentioned"
 
+    async def test_handle_passive_message_response_policy_skips_response_but_still_extracts(
+        self, mock_provider, mock_agent, caplog
+    ):
+        """Response policy should suppress passive replies while still extracting."""
+        import asyncio
+        import logging
+
+        from ash.config.models import PassiveListeningConfig
+        from ash.providers.base import IncomingMessage
+        from ash.providers.telegram.handlers import TelegramMessageHandler
+
+        mock_provider.passive_config = PassiveListeningConfig(
+            enabled=True,
+            response_allowed_chats=["group_allowed"],
+        )
+        mock_provider.bot_username = "ash_bot"
+
+        handler = TelegramMessageHandler(
+            provider=mock_provider,
+            agent=mock_agent,
+            streaming=False,
+        )
+
+        mock_decider = MagicMock()
+        mock_decider.decide = AsyncMock(return_value=True)
+        handler._passive_handler._passive_decider = mock_decider  # type: ignore[union-attr]
+        handler._passive_handler._memory_manager = MagicMock()  # type: ignore[union-attr]
+        handler._passive_handler._passive_extractor = MagicMock()  # type: ignore[union-attr]
+        handler._passive_handler._extract_passive_memories = AsyncMock()  # type: ignore[method-assign]
+
+        passive_message = IncomingMessage(
+            id="99",
+            chat_id="group_blocked",
+            user_id="user_456",
+            text="normal group chatter",
+            username="otheruser",
+            display_name="Other User",
+        )
+
+        with caplog.at_level(logging.INFO, logger="telegram"):
+            await handler.handle_passive_message(passive_message)
+        await asyncio.sleep(0)
+
+        mock_decider.decide.assert_not_called()
+        handler._passive_handler._extract_passive_memories.assert_awaited_once_with(  # type: ignore[attr-defined]
+            passive_message
+        )
+        mock_agent.process_message.assert_not_called()
+        mock_agent.process_message_streaming.assert_not_called()
+        skipped = next(
+            (r for r in caplog.records if r.msg == "passive_engagement_skipped"), None
+        )
+        assert skipped is not None
+        assert getattr(skipped, "decision_path", None) == "response_policy"
+        assert getattr(skipped, "engagement_reason", None) == "response_policy"
+
     async def test_handle_passive_message_direct_followup_bypasses_throttle_but_uses_decider(
         self, mock_provider, mock_agent, tmp_path, caplog
     ):