feat(gateway): per-platform typing_indicator toggle

Add a generic per-platform PlatformConfig.typing_indicator flag (default
True) that gates the _keep_typing refresh loop in
_process_message_background. When false, the loop is never spawned, so no
typing/"is thinking…" status is shown on that platform — message delivery
is otherwise unchanged.

Mirrors the gateway_restart_notification contract exactly: dataclass field
+ to_dict/from_dict (with extra-fallback resolution) + shared-key bridge in
load_gateway_config, so 'slack: typing_indicator: false' under platforms
works without a separate block. Generic by design — the same key works for
every platform (Slack 'is thinking…', Telegram/Discord/Signal typing).

Motivated by users who find Slack's assistant 'is thinking…' status noisy
(it also briefly disables the compose box, via the Assistant API).

gateway/config.py

@@ -345,6 +345,15 @@ class PlatformConfig:
     # noise; keep True for back-channels where the operator wants them.
     gateway_restart_notification: bool = True
 
+    # Whether the gateway shows a "typing…" / "is thinking…" status indicator
+    # while the agent processes a message on this platform. Default True
+    # preserves prior behavior. Set False on platforms where the indicator is
+    # unwanted (e.g. Slack's assistant.threads.setStatus "is thinking…", which
+    # disables the compose box, or any platform where users find the bubble
+    # noisy). Drives the per-message _keep_typing refresh loop in
+    # gateway/platforms/base.py.
+    typing_indicator: bool = True
+
     # Platform-specific settings
     extra: Dict[str, Any] = field(default_factory=dict)
 
@@ -354,6 +363,7 @@ class PlatformConfig:
             "extra": self.extra,
             "reply_to_mode": self.reply_to_mode,
             "gateway_restart_notification": self.gateway_restart_notification,
+            "typing_indicator": self.typing_indicator,
         }
         if self.token:
             result["token"] = self.token
@@ -377,6 +387,13 @@ class PlatformConfig:
         if _grn is None:
             _grn = data.get("extra", {}).get("gateway_restart_notification")
 
+        # typing_indicator mirrors gateway_restart_notification: it may arrive
+        # top-level or bridged into extra by the shared-key loop in
+        # load_gateway_config(), so check both.
+        _typing = data.get("typing_indicator")
+        if _typing is None:
+            _typing = data.get("extra", {}).get("typing_indicator")
+
         return cls(
             enabled=_coerce_bool(data.get("enabled"), False),
             token=data.get("token"),
@@ -384,6 +401,7 @@ class PlatformConfig:
             home_channel=home_channel,
             reply_to_mode=data.get("reply_to_mode", "first"),
             gateway_restart_notification=_coerce_bool(_grn, True),
+            typing_indicator=_coerce_bool(_typing, True),
             extra=data.get("extra", {}),
         )
 
@@ -1032,6 +1050,8 @@ def load_gateway_config() -> GatewayConfig:
                         bridged["channel_prompts"] = channel_prompts
                 if "gateway_restart_notification" in platform_cfg:
                     bridged["gateway_restart_notification"] = platform_cfg["gateway_restart_notification"]
+                if "typing_indicator" in platform_cfg:
+                    bridged["typing_indicator"] = platform_cfg["typing_indicator"]
                 enabled_was_explicit = _cfg_toplevel and "enabled" in platform_cfg
                 if not bridged and not enabled_was_explicit:
                     continue

gateway/platforms/base.py

@@ -4562,21 +4562,26 @@ class BasePlatformAdapter(ABC):
         interrupt_event = self._active_sessions.get(session_key) or asyncio.Event()
         self._active_sessions[session_key] = interrupt_event
         
-        # Start continuous typing indicator (refreshes every 2 seconds)
+        # Start continuous typing indicator (refreshes every 2 seconds).
+        # Gated per-platform: when typing_indicator=False the refresh loop is
+        # never spawned, so no "typing…" / "is thinking…" status is shown.
+        # typing_task stays None; _stop_typing_refresh already no-ops on None.
         _thread_metadata = _thread_metadata_for_source(event.source, _reply_anchor_for_event(event))
-        _keep_typing_kwargs = {"metadata": _thread_metadata}
-        try:
-            _keep_typing_sig = inspect.signature(self._keep_typing)
-        except (TypeError, ValueError):
-            _keep_typing_sig = None
-        if _keep_typing_sig is None or "stop_event" in _keep_typing_sig.parameters:
-            _keep_typing_kwargs["stop_event"] = interrupt_event
-        typing_task = asyncio.create_task(
-            self._keep_typing(
-                event.source.chat_id,
-                **_keep_typing_kwargs,
+        typing_task: Optional[asyncio.Task] = None
+        if getattr(self.config, "typing_indicator", True):
+            _keep_typing_kwargs: Dict[str, Any] = {"metadata": _thread_metadata}
+            try:
+                _keep_typing_sig = inspect.signature(self._keep_typing)
+            except (TypeError, ValueError):
+                _keep_typing_sig = None
+            if _keep_typing_sig is None or "stop_event" in _keep_typing_sig.parameters:
+                _keep_typing_kwargs["stop_event"] = interrupt_event
+            typing_task = asyncio.create_task(
+                self._keep_typing(
+                    event.source.chat_id,
+                    **_keep_typing_kwargs,
+                )
             )
-        )
 
         async def _stop_typing_task() -> None:
             await self._stop_typing_refresh(

tests/gateway/test_config.py

@@ -71,6 +71,25 @@ class TestPlatformConfigRoundtrip:
         restored = PlatformConfig.from_dict({"gateway_restart_notification": "false"})
         assert restored.gateway_restart_notification is False
 
+    def test_typing_indicator_defaults_true(self):
+        assert PlatformConfig().typing_indicator is True
+        assert PlatformConfig.from_dict({}).typing_indicator is True
+
+    def test_typing_indicator_roundtrip_false(self):
+        pc = PlatformConfig(enabled=True, typing_indicator=False)
+        restored = PlatformConfig.from_dict(pc.to_dict())
+        assert restored.typing_indicator is False
+
+    def test_typing_indicator_coerces_quoted_false(self):
+        restored = PlatformConfig.from_dict({"typing_indicator": "false"})
+        assert restored.typing_indicator is False
+
+    def test_typing_indicator_resolved_from_extra(self):
+        # The shared-key loop in load_gateway_config bridges the flag into
+        # extra; from_dict must honor it there too (mirrors _grn fallback).
+        restored = PlatformConfig.from_dict({"extra": {"typing_indicator": False}})
+        assert restored.typing_indicator is False
+
 
 class TestGetConnectedPlatforms:
     def test_returns_enabled_with_token(self):

tests/gateway/test_typing_indicator_toggle.py

@@ -0,0 +1,99 @@
+"""Per-platform typing-indicator toggle (PlatformConfig.typing_indicator).
+
+The "typing…" / "is thinking…" status bubble is driven by the generic
+``_keep_typing`` refresh loop that ``_process_message_background`` spawns for
+every inbound message on every platform.  ``typing_indicator`` (default True)
+gates that spawn: when False, the loop is never started, so ``send_typing``
+is never called and no status indicator is shown — while message delivery is
+otherwise unchanged.
+
+These are behavioral tests against the real dispatch path, not snapshots.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+)
+from gateway.session import SessionSource, build_session_key
+
+
+class _StubAdapter(BasePlatformAdapter):
+    async def connect(self, *, is_reconnect: bool = False):
+        pass
+
+    async def disconnect(self):
+        pass
+
+    async def send(self, chat_id, text, **kwargs):
+        return None
+
+    async def get_chat_info(self, chat_id):
+        return {}
+
+
+def _make_adapter(typing_indicator: bool) -> _StubAdapter:
+    adapter = _StubAdapter(
+        PlatformConfig(enabled=True, token="t", typing_indicator=typing_indicator),
+        Platform.SLACK,
+    )
+    # Record send_typing calls without performing any platform I/O.
+    adapter.send_typing = AsyncMock(return_value=None)
+    adapter._send_with_retry = AsyncMock(return_value=None)
+    # Handler returns immediately; the typing loop only fires if it was spawned.
+    adapter._message_handler = AsyncMock(return_value="ok")
+    return adapter
+
+
+def _make_event(chat_id="C123"):
+    return MessageEvent(
+        text="hi",
+        message_type=MessageType.TEXT,
+        source=SessionSource(platform=Platform.SLACK, chat_id=chat_id, chat_type="dm"),
+    )
+
+
+def _sk(chat_id="C123"):
+    return build_session_key(
+        SessionSource(platform=Platform.SLACK, chat_id=chat_id, chat_type="dm")
+    )
+
+
+@pytest.mark.asyncio
+async def test_typing_indicator_enabled_spawns_refresh_loop():
+    """Default (typing_indicator=True): the refresh loop calls send_typing."""
+    adapter = _make_adapter(typing_indicator=True)
+
+    # Real handlers take time (tool calls); yield long enough for the spawned
+    # refresh loop to fire at least one send_typing before delivery completes.
+    async def _slow_handler(_event):
+        await asyncio.sleep(0.05)
+        return "ok"
+
+    adapter._message_handler = _slow_handler
+    event = _make_event()
+    adapter._active_sessions[_sk()] = asyncio.Event()
+
+    await adapter._process_message_background(event, _sk())
+
+    assert adapter.send_typing.await_count >= 1
+
+
+@pytest.mark.asyncio
+async def test_typing_indicator_disabled_never_calls_send_typing():
+    """typing_indicator=False: the loop is never spawned, send_typing unused."""
+    adapter = _make_adapter(typing_indicator=False)
+    event = _make_event()
+    adapter._active_sessions[_sk()] = asyncio.Event()
+
+    await adapter._process_message_background(event, _sk())
+
+    adapter.send_typing.assert_not_awaited()
+    # Delivery still happened — disabling typing must not suppress the reply.
+    adapter._send_with_retry.assert_awaited()

website/docs/user-guide/messaging/index.md

@@ -579,6 +579,21 @@ gateway:
 
 Disable it on noisy or low-priority platforms while leaving it on for your primary chat. The notification is sent once per restart, regardless of how many sessions were in flight.
 
+### Typing indicators
+
+While the agent is processing a message, the gateway shows a live typing status on platforms that support it — a "typing…" bubble on Telegram/Discord/Signal, or the "is thinking…" assistant status on Slack. This is controlled per-platform by the `typing_indicator` flag in `gateway-config.yaml`, which defaults to `true`:
+
+```yaml
+gateway:
+  platforms:
+    slack:
+      typing_indicator: false   # don't show "is thinking…" on Slack
+    telegram:
+      # typing_indicator omitted → defaults to true
+```
+
+Set `typing_indicator: false` on any platform where the indicator is unwanted. Some users find Slack's "is thinking…" status noisy (it also briefly disables the compose box while shown, since it uses Slack's Assistant API). Disabling it only suppresses the indicator — message delivery and everything else is unchanged. The flag is generic, so the same key works for every platform.
+
 ### Session resume across gateway restarts
 
 When the gateway shuts down with an in-flight tool call or generation, the affected sessions are flagged as `restart_interrupted`. On the next startup, the gateway schedules an auto-resume for each one — the user gets a short heads-up in the chat ("Send any message after restart and I'll try to resume where you left off.") and the session picks up from the last committed turn when they reply.

website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/messaging/index.md

@@ -495,6 +495,21 @@ gateway:
 
 在嘈杂或低优先级的平台上禁用，同时在主要聊天上保持启用。无论有多少会话正在进行，每次重启只发送一次通知。
 
+### 正在输入指示器
+
+当 agent 正在处理消息时，网关会在支持的平台上显示实时的输入状态——Telegram/Discord/Signal 上的"正在输入……"气泡，或 Slack 上的"is thinking…"助手状态。这由 `gateway-config.yaml` 中每个平台的 `typing_indicator` 标志控制，默认为 `true`：
+
+```yaml
+gateway:
+  platforms:
+    slack:
+      typing_indicator: false   # 在 Slack 上不显示"is thinking…"
+    telegram:
+      # typing_indicator 未设置 → 默认为 true
+```
+
+在任何不需要该指示器的平台上设置 `typing_indicator: false`。部分用户觉得 Slack 的"is thinking…"状态比较嘈杂（由于它使用 Slack 的 Assistant API，显示期间还会短暂禁用输入框）。禁用它只会抑制该指示器——消息投递及其他一切均不受影响。该标志是通用的，因此同一个键对每个平台都有效。
+
 ### 网关重启后的会话恢复
 
 当网关在工具调用或生成进行中时关闭，受影响的会话被标记为 `restart_interrupted`。下次启动时，网关为每个会话安排自动恢复——用户在聊天中收到简短提示（"Send any message after restart and I'll try to resume where you left off."），当他们回复时，会话从最后提交的轮次继续。