feat(connector): add reply visibility policy

connector/platforms/feishu_webhook.py

@@ -25,6 +25,7 @@ from urllib.parse import urlparse
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest
+from ..reply_visibility import decide_reply_visibility
 from ..router import CommandRouter
 from ..security_profile import AllowlistPolicy, ReplayGuard
 from .feishu_installation_manager import FeishuBinding, FeishuInstallationManager
@@ -638,6 +639,7 @@ class FeishuWebhookServer:
             metadata={
                 "account_id": binding.account_id,
                 "chat_type": chat_type,
+                "mentioned_bot": mentioned_bot,
                 "message_type": str(message.get("message_type", "") or "").strip(),
                 "sender_open_id": sender_open_id,
             },
@@ -690,7 +692,17 @@ class FeishuWebhookServer:
         if buttons:
             await self._send_interactive_reply(target, resp_text, buttons)
         elif resp_text:
-            await self._send_reply(target, resp_text)
+            await self._send_reply(
+                target,
+                resp_text,
+                delivery_context={
+                    "workspace_id": request.workspace_id,
+                    "thread_id": request.thread_id,
+                    "account_id": str(request.metadata.get("account_id", "") or ""),
+                    "chat_type": str(request.metadata.get("chat_type", "") or ""),
+                    "mentioned_bot": bool(request.metadata.get("mentioned_bot")),
+                },
+            )
 
     def _resolve_delivery_binding(
         self, *, workspace_id: str = "", account_id: str = ""
@@ -1110,7 +1122,34 @@ class FeishuWebhookServer:
                 "Feishu interactive reply failed: %s", data.get("msg", "unknown")
             )
 
-    async def _send_reply(self, target: FeishuDeliveryTarget, text: str) -> None:
+    async def _send_reply(
+        self,
+        target: FeishuDeliveryTarget,
+        text: str,
+        *,
+        delivery_context: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        ctx = dict(delivery_context or {})
+        if target.workspace_id:
+            ctx.setdefault("workspace_id", target.workspace_id)
+        if target.account_id:
+            ctx.setdefault("account_id", target.account_id)
+        if target.reply_to_message_id:
+            ctx.setdefault("thread_id", target.reply_to_message_id)
+        decision = decide_reply_visibility(
+            delivery_context=ctx,
+            platform="feishu",
+            channel_kind=str(ctx.get("chat_type", "") or ""),
+            in_thread=bool(target.reply_to_message_id),
+            text=text,
+        )
+        if decision.suppressed:
+            logger.info(
+                "Suppressed Feishu reply channel=%s reason=%s",
+                target.channel_id,
+                decision.reason,
+            )
+            return
         resolution, binding, _ = self._resolve_delivery_binding(
             workspace_id=target.workspace_id,
             account_id=target.account_id,
@@ -1193,6 +1232,7 @@ class FeishuWebhookServer:
                 account_id=str(ctx.get("account_id", "") or "").strip(),
             ),
             text,
+            delivery_context=ctx,
         )
 
     async def send_image(

connector/platforms/slack_webhook.py

@@ -38,6 +38,7 @@ from urllib.parse import parse_qs
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest, CommandResponse
+from ..reply_visibility import decide_reply_visibility
 from ..router import CommandRouter
 from ..security_profile import AllowlistPolicy, ReplayGuard
 from .slack_installation_manager import SlackInstallationManager
@@ -54,6 +55,12 @@ _SLACK_INTERACTION_TYPES = frozenset(
 )
 
 
+def _slack_channel_kind(channel_id: str) -> str:
+    if str(channel_id or "").startswith("D"):
+        return "dm"
+    return "group"
+
+
 # -- aiohttp compat layer (same pattern as kakao/whatsapp/wechat) -----------
 
 
@@ -606,6 +613,9 @@ class SlackWebhookServer:
 
         # S67: Require mention in group channels.
         is_dm = channel_id.startswith("D")
+        mentioned_bot = event_type == "app_mention" or (
+            bool(bot_user_id) and f"<@{bot_user_id}>" in text
+        )
         if not is_dm and self.config.slack_require_mention:
             if event_type != "app_mention":
                 if bot_user_id and f"<@{bot_user_id}>" not in text:
@@ -658,6 +668,8 @@ class SlackWebhookServer:
                         delivery_context={
                             "workspace_id": workspace_id,
                             "thread_id": req.thread_id,
+                            "channel_kind": _slack_channel_kind(channel_id),
+                            "mentioned": mentioned_bot,
                         },
                     )
                 else:
@@ -668,6 +680,8 @@ class SlackWebhookServer:
                         delivery_context={
                             "workspace_id": workspace_id,
                             "thread_id": req.thread_id,
+                            "channel_kind": _slack_channel_kind(channel_id),
+                            "mentioned": mentioned_bot,
                         },
                     )
         except Exception as e:
@@ -1039,15 +1053,29 @@ class SlackWebhookServer:
         delivery_context: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Send a message via Slack Web API (chat.postMessage)."""
+        ctx = dict(delivery_context or {})
+        if not thread_ts:
+            thread_ts = str(ctx.get("thread_id", "") or "").strip()
+        decision = decide_reply_visibility(
+            delivery_context=ctx,
+            platform="slack",
+            channel_kind=_slack_channel_kind(channel_id),
+            in_thread=bool(thread_ts),
+            text=text,
+        )
+        if decision.suppressed:
+            logger.info(
+                "Suppressed Slack reply channel=%s reason=%s",
+                channel_id,
+                decision.reason,
+            )
+            return
         try:
             import aiohttp as _aiohttp
         except ImportError:
             logger.warning("aiohttp not available; cannot send Slack reply")
             return
 
-        ctx = dict(delivery_context or {})
-        if not thread_ts:
-            thread_ts = str(ctx.get("thread_id", "") or "").strip()
         installation_id, bot_token, workspace_id = self._resolve_workspace_credentials(
             str(ctx.get("workspace_id", "") or "").strip()
         )

connector/platforms/telegram_polling.py

@@ -13,6 +13,7 @@ from services.connector_replay_lifecycle import ConnectorReplayLifecycle
 
 from ..config import ConnectorConfig
 from ..contract import CommandRequest, CommandResponse
+from ..reply_visibility import decide_reply_visibility
 from ..router import CommandRouter
 from ..state import ConnectorState
 
@@ -43,6 +44,15 @@ def _normalize_message_thread_id(value) -> Optional[int]:
     return thread_id
 
 
+def _telegram_channel_kind(chat_id) -> str:
+    text = str(chat_id or "").strip()
+    if text.startswith("-100"):
+        return "supergroup"
+    if text.startswith("-"):
+        return "group"
+    return "dm"
+
+
 class TelegramPolling:
     def __init__(self, config: ConnectorConfig, router: CommandRouter):
         self.config = config
@@ -255,6 +265,21 @@ class TelegramPolling:
         resp: CommandResponse,
         delivery_context: Optional[dict] = None,
     ) -> bool:
+        decision = decide_reply_visibility(
+            delivery_context=dict(delivery_context or {}),
+            platform="telegram",
+            channel_kind=_telegram_channel_kind(chat_id),
+            text=getattr(resp, "text", ""),
+            has_buttons=bool(getattr(resp, "buttons", None)),
+            has_files=bool(getattr(resp, "files", None)),
+        )
+        if decision.suppressed:
+            logger.info(
+                "Suppressed Telegram reply chat=%s reason=%s",
+                chat_id,
+                decision.reason,
+            )
+            return True
         url = f"{self.base_url}/sendMessage"
         payload = {
             "chat_id": chat_id,
@@ -326,6 +351,19 @@ class TelegramPolling:
         """Send text message."""
         if not self.session:
             return
+        decision = decide_reply_visibility(
+            delivery_context=dict(delivery_context or {}),
+            platform="telegram",
+            channel_kind=_telegram_channel_kind(channel_id),
+            text=text,
+        )
+        if decision.suppressed:
+            logger.info(
+                "Suppressed Telegram send_message chat=%s reason=%s",
+                channel_id,
+                decision.reason,
+            )
+            return
 
         # Reuse internal logic logic but public
         # Using simplified direct call

connector/reply_visibility.py

@@ -0,0 +1,129 @@
+"""Shared connector reply visibility decisions."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+VISIBLE = "visible"
+SUPPRESS_TEXT = "suppress_text"
+TOOL_ONLY = "tool_only"
+INTERNAL = "internal"
+AUTO = "auto"
+
+_VISIBLE_VALUES = {"", AUTO, VISIBLE, "public", "reply", "send"}
+_SUPPRESS_VALUES = {SUPPRESS_TEXT, "suppress", "silent", "no_text", "none"}
+_TOOL_ONLY_VALUES = {TOOL_ONLY, "tool-only", "tool", "action_only", "action-only"}
+_INTERNAL_VALUES = {INTERNAL, "internal_only", "internal-only", "private"}
+_TRUTHY_VALUES = {"1", "true", "yes", "y", "on"}
+
+
+@dataclass(frozen=True)
+class ReplyVisibilityDecision:
+    visible: bool
+    mode: str
+    reason: str
+    diagnostics: Dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def suppressed(self) -> bool:
+        return not self.visible
+
+
+def normalize_reply_visibility_mode(value: Any) -> str:
+    text = str(value or "").strip().lower()
+    if text in _VISIBLE_VALUES:
+        return VISIBLE
+    if text in _SUPPRESS_VALUES:
+        return SUPPRESS_TEXT
+    if text in _TOOL_ONLY_VALUES:
+        return TOOL_ONLY
+    if text in _INTERNAL_VALUES:
+        return INTERNAL
+    return VISIBLE
+
+
+def decide_reply_visibility(
+    *,
+    delivery_context: Optional[Dict[str, Any]] = None,
+    platform: str = "",
+    channel_kind: str = "",
+    mentioned: Optional[bool] = None,
+    in_thread: bool = False,
+    text: str = "",
+    has_buttons: bool = False,
+    has_files: bool = False,
+) -> ReplyVisibilityDecision:
+    ctx = dict(delivery_context or {})
+    explicit_mode = _extract_mode(ctx)
+    mode = normalize_reply_visibility_mode(explicit_mode)
+    normalized_channel_kind = (
+        str(ctx.get("channel_kind") or ctx.get("chat_type") or channel_kind or "")
+        .strip()
+        .lower()
+    )
+    threaded = bool(in_thread or str(ctx.get("thread_id", "") or "").strip())
+
+    diagnostics = {
+        "platform": str(platform or ctx.get("platform", "") or "").strip(),
+        "mode": mode,
+        "channel_kind": normalized_channel_kind,
+        "in_thread": threaded,
+        "has_text": bool(str(text or "").strip()),
+        "has_buttons": bool(has_buttons),
+        "has_files": bool(has_files),
+    }
+
+    # Approval/action replies must stay visible; hiding them can strand operators.
+    if has_buttons:
+        return ReplyVisibilityDecision(
+            True, VISIBLE, "interactive_action_required", diagnostics
+        )
+
+    if mode == INTERNAL or _truthy(ctx.get("internal_delivery")):
+        return ReplyVisibilityDecision(
+            False, INTERNAL, "internal_delivery", diagnostics
+        )
+
+    if (
+        mode in {SUPPRESS_TEXT, TOOL_ONLY}
+        or _truthy(ctx.get("tool_only"))
+        or _truthy(ctx.get("silent"))
+    ):
+        if has_files:
+            return ReplyVisibilityDecision(
+                True, VISIBLE, "file_delivery_preserved", diagnostics
+            )
+        return ReplyVisibilityDecision(
+            False, mode, "text_reply_suppressed", diagnostics
+        )
+
+    if normalized_channel_kind in {"group", "supergroup", "channel"}:
+        if mentioned is None and "mentioned" in ctx:
+            mentioned = _truthy(ctx.get("mentioned"))
+        if mentioned is None and "mentioned_bot" in ctx:
+            mentioned = _truthy(ctx.get("mentioned_bot"))
+        if mentioned is False and not threaded:
+            return ReplyVisibilityDecision(
+                False, SUPPRESS_TEXT, "group_no_mention", diagnostics
+            )
+
+    return ReplyVisibilityDecision(True, VISIBLE, "visible", diagnostics)
+
+
+def _extract_mode(ctx: Dict[str, Any]) -> Any:
+    for key in ("reply_visibility", "visibility", "reply_visibility_mode"):
+        if key in ctx:
+            return ctx.get(key)
+    policy = ctx.get("delivery_policy")
+    if isinstance(policy, dict):
+        for key in ("reply_visibility", "visibility", "reply_visibility_mode"):
+            if key in policy:
+                return policy.get(key)
+    return AUTO
+
+
+def _truthy(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    return str(value or "").strip().lower() in _TRUTHY_VALUES

connector/results_poller.py

@@ -6,6 +6,7 @@ from typing import Any, Dict, Optional
 from .config import ConnectorConfig
 from .contract import Platform
 from .openclaw_client import OpenClawClient
+from .reply_visibility import decide_reply_visibility
 
 logger = logging.getLogger(__name__)
 
@@ -363,13 +364,28 @@ class ResultsPoller:
         *,
         delivery_context: Optional[Dict[str, Any]] = None,
     ):
+        ctx = dict(delivery_context or {})
+        decision = decide_reply_visibility(
+            delivery_context=ctx,
+            platform=platform_name,
+            text=text,
+        )
+        if decision.suppressed:
+            # IMPORTANT: a suppressed visible reply is a successful delivery no-op.
+            logger.info(
+                "Suppressed connector text reply platform=%s channel=%s reason=%s",
+                platform_name,
+                channel_id,
+                decision.reason,
+            )
+            return
         platform = self.platforms.get(platform_name)
         if platform:
             try:
                 await platform.send_message(
                     channel_id,
                     text,
-                    delivery_context=dict(delivery_context or {}),
+                    delivery_context=ctx,
                 )
             except Exception as e:
                 logger.error(f"Failed to send text to {platform_name}: {e}")

tests/test_f74_reply_visibility_policy.py

@@ -0,0 +1,206 @@
+import sys
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from connector.config import ConnectorConfig
+from connector.contract import CommandResponse, Platform
+from connector.platforms.feishu_webhook import FeishuWebhookServer
+from connector.platforms.slack_webhook import SlackWebhookServer
+from connector.platforms.telegram_polling import TelegramPolling
+from connector.reply_visibility import decide_reply_visibility
+from connector.results_poller import ResultsPoller
+
+
+class _FakeResponse:
+    status = 200
+
+    async def text(self):
+        return "OK"
+
+    async def json(self):
+        return {"ok": True}
+
+
+class _FakePostContext:
+    async def __aenter__(self):
+        return _FakeResponse()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+
+class _FakeSession:
+    def __init__(self):
+        self.posts = []
+
+    def post(self, url, **kwargs):
+        self.posts.append((url, kwargs))
+        return _FakePostContext()
+
+
+class _MockPlatform(Platform):
+    async def send_image(
+        self,
+        channel_id,
+        image_data,
+        filename="image.png",
+        caption=None,
+        delivery_context=None,
+    ):
+        pass
+
+    async def send_message(self, channel_id, text, delivery_context=None):
+        pass
+
+
+class TestF74ReplyVisibilityPolicy(unittest.IsolatedAsyncioTestCase):
+    def test_shared_policy_matrix(self):
+        cases = [
+            (
+                "dm",
+                decide_reply_visibility(
+                    delivery_context={"chat_type": "p2p"}, platform="feishu", text="ok"
+                ),
+                True,
+                "visible",
+            ),
+            (
+                "group mention",
+                decide_reply_visibility(
+                    delivery_context={"chat_type": "group", "mentioned_bot": True},
+                    platform="feishu",
+                    text="ok",
+                ),
+                True,
+                "visible",
+            ),
+            (
+                "group no mention",
+                decide_reply_visibility(
+                    delivery_context={"chat_type": "group", "mentioned_bot": False},
+                    platform="feishu",
+                    text="ok",
+                ),
+                False,
+                "group_no_mention",
+            ),
+            (
+                "thread",
+                decide_reply_visibility(
+                    delivery_context={
+                        "chat_type": "group",
+                        "mentioned_bot": False,
+                        "thread_id": "t-1",
+                    },
+                    platform="slack",
+                    text="ok",
+                ),
+                True,
+                "visible",
+            ),
+            (
+                "internal",
+                decide_reply_visibility(
+                    delivery_context={"internal_delivery": True},
+                    platform="telegram",
+                    text="ok",
+                ),
+                False,
+                "internal_delivery",
+            ),
+            (
+                "tool only",
+                decide_reply_visibility(
+                    delivery_context={"reply_visibility": "tool_only"},
+                    platform="telegram",
+                    text="ok",
+                ),
+                False,
+                "text_reply_suppressed",
+            ),
+            (
+                "interactive",
+                decide_reply_visibility(
+                    delivery_context={"reply_visibility": "tool_only"},
+                    platform="slack",
+                    text="approval",
+                    has_buttons=True,
+                ),
+                True,
+                "interactive_action_required",
+            ),
+        ]
+        for label, decision, visible, reason in cases:
+            with self.subTest(label=label):
+                self.assertEqual(decision.visible, visible)
+                self.assertEqual(decision.reason, reason)
+
+    async def test_result_poller_suppressed_text_is_successful_noop(self):
+        config = ConnectorConfig()
+        client = MagicMock()
+        platform = _MockPlatform()
+        platform.send_message = AsyncMock(side_effect=AssertionError("must not send"))
+        platform.send_image = AsyncMock()
+        poller = ResultsPoller(config, client, {"test": platform})
+
+        await poller._deliver_results(
+            "p-quiet",
+            {"outputs": {}},
+            "test",
+            "c-1",
+            delivery_context={"reply_visibility": "tool_only"},
+        )
+
+        platform.send_message.assert_not_called()
+        platform.send_image.assert_not_called()
+
+
+class TestF74ReplyVisibilityAdapters(unittest.IsolatedAsyncioTestCase):
+    async def test_telegram_suppressed_response_returns_success_without_http_send(self):
+        config = ConnectorConfig()
+        config.telegram_bot_token = "telegram-token"
+        server = TelegramPolling(config, MagicMock())
+        server.session = _FakeSession()
+
+        delivered = await server._send_response(
+            -100123,
+            CommandResponse(text="quiet"),
+            delivery_context={"reply_visibility": "tool_only"},
+        )
+
+        self.assertTrue(delivered)
+        self.assertEqual(server.session.posts, [])
+
+    async def test_slack_suppressed_reply_skips_chat_post(self):
+        config = ConnectorConfig()
+        config.slack_bot_token = "slack-token"
+        server = SlackWebhookServer(config, MagicMock())
+
+        fake_aiohttp = MagicMock()
+        with patch.dict(sys.modules, {"aiohttp": fake_aiohttp}):
+            await server._send_reply(
+                channel_id="C_F74",
+                text="quiet",
+                delivery_context={"reply_visibility": "tool_only"},
+            )
+
+        fake_aiohttp.ClientSession.assert_not_called()
+
+    async def test_feishu_suppressed_reply_skips_open_api_send(self):
+        config = ConnectorConfig()
+        server = FeishuWebhookServer(config, MagicMock())
+
+        with patch(
+            "connector.platforms.feishu_webhook.safe_request_json"
+        ) as mock_safe_request:
+            await server.send_message(
+                "oc_group_1",
+                "quiet",
+                delivery_context={"reply_visibility": "tool_only"},
+            )
+
+        mock_safe_request.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()