fix(gateway): emit silent lifecycle activity without progress bubbles

gateway/platforms/base.py

@@ -5302,6 +5302,15 @@ def pop_post_delivery_callback(
     async def on_processing_start(self, event: MessageEvent) -> None:
         """Hook called when background processing begins."""
 
+    async def on_processing_activity(
+        self,
+        source: SessionSource,
+        message_id: str,
+        event_type: str,
+        tool_name: Optional[str] = None,
+    ) -> None:
+        """Hook called when the active reasoning/tool state changes."""
+
     async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None:
         """Hook called when background processing completes.
 

gateway/run.py

@@ -3839,6 +3839,24 @@ def progress_callback(self, event_type: str, tool_name: str = None, preview: str
                     ctx._live_status_adapter.set_status_text(ctx.source.chat_id, None)
             except Exception as _ls_err:
                 logger.debug("live status update failed: %s", _ls_err)
+
+        # Platform lifecycle reactions/status are independent of visible
+        # progress messages. Adapters can reflect state without creating a
+        # second chat surface or notification.
+        if ctx._activity_enabled and ctx._run_still_current():
+            safe_schedule_threadsafe(
+                ctx._activity_adapter._run_processing_hook(
+                    "on_processing_activity",
+                    ctx.source,
+                    ctx.event_message_id,
+                    event_type,
+                    tool_name,
+                ),
+                ctx._activity_loop,
+                logger=logger,
+                log_message="processing activity hook scheduling error",
+            )
+
         # "log" mode: append tool.started lines to the log queue and stay
         # silent in chat. Handled before the progress_queue guard because
         # log mode runs without a chat progress queue.
@@ -5042,6 +5060,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                 ctx.needs_progress_queue
                 or ctx.log_mode_enabled
                 or ctx._live_status_adapter is not None
+                or ctx._activity_enabled
             )
             else None
         )
@@ -25577,6 +25596,16 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview:
             _cleanup_progress = False
             _cleanup_adapter = None
         _cleanup_msg_ids: List[str] = []
+        _activity_adapter = self._adapter_for_source(source)
+        # Activity hooks target the original inbound message. A turn without
+        # that platform reference has no safe status surface and remains a no-op.
+        _activity_enabled = bool(
+            _activity_adapter is not None
+            and event_message_id
+            and type(_activity_adapter).on_processing_activity
+            is not BasePlatformAdapter.on_processing_activity
+        )
+        _activity_loop = asyncio.get_running_loop()
         # First-touch onboarding latch: fires at most once per run, even if
         # several tools exceed the threshold.
         long_tool_hint_fired = [False]
@@ -25587,6 +25616,9 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview:
             _run_still_current=_run_still_current,
             _live_status_adapter=_live_status_adapter,
             _live_status_mode=_live_status_mode,
+            _activity_adapter=_activity_adapter,
+            _activity_enabled=_activity_enabled,
+            _activity_loop=_activity_loop,
             _thinking_enabled=_thinking_enabled,
             progress_mode=progress_mode,
             progress_grouping=progress_grouping,

gateway/turn_context.py

@@ -38,6 +38,9 @@ class TurnContext:
     _run_still_current: Callable[[], bool] = None  # type: ignore[assignment]
     _live_status_adapter: Any = None
     _live_status_mode: str = "off"
+    _activity_adapter: Any = None
+    _activity_enabled: bool = False
+    _activity_loop: Any = None
     _thinking_enabled: bool = False
     progress_mode: str = "off"
     progress_grouping: str = "grouped"

plugins/platforms/telegram/adapter.py

@@ -9927,41 +9927,130 @@ def _reactions_enabled(self) -> bool:
         """Check if message reactions are enabled via config/env."""
         return os.getenv("TELEGRAM_REACTIONS", "false").lower() not in {"false", "0", "no"}
 
-    async def _set_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
-        """Set a single emoji reaction on a Telegram message."""
-        if not self._bot:
-            return False
-        try:
-            await self._bot.set_message_reaction(
-                chat_id=normalize_telegram_chat_id(chat_id),
-                message_id=int(message_id),
-                reaction=emoji,
-            )
-            return True
-        except Exception as e:
-            logger.debug("[%s] set_message_reaction failed (%s): %s", self.name, emoji, _redact_telegram_error_text(e))
-            return False
+    def _reaction_update_lock(self) -> asyncio.Lock:
+        """Serialize reaction updates so cache and Telegram stay in sync."""
+        lock = getattr(self, "_processing_reactions_lock", None)
+        if lock is None:
+            lock = self._processing_reactions_lock = asyncio.Lock()
+        return lock
 
-    async def _clear_reactions(self, chat_id: str, message_id: str) -> bool:
+    def _settle_reaction_key(self, key: tuple[str, str]) -> None:
+        """Remember a terminal message briefly so late activity cannot overwrite it."""
+        settled = getattr(self, "_settled_processing_reactions", None)
+        if settled is None:
+            settled = self._settled_processing_reactions = {}
+        settled.pop(key, None)
+        settled[key] = None
+        if len(settled) > 1024:
+            settled.pop(next(iter(settled)))
+
+    async def _set_reaction(
+        self,
+        chat_id: str,
+        message_id: str,
+        emoji: str,
+        *,
+        retain_state: bool = True,
+        begin_state: bool = False,
+        activity_state: bool = False,
+        terminal_state: bool = False,
+    ) -> bool:
+        """Set one reaction, skipping duplicate state updates."""
+        key = (str(chat_id), str(message_id))
+        async with self._reaction_update_lock():
+            states = getattr(self, "_processing_reactions", None)
+            if states is None:
+                states = self._processing_reactions = {}
+            settled = getattr(self, "_settled_processing_reactions", None)
+            if begin_state and settled is not None:
+                settled.pop(key, None)
+            elif activity_state and settled is not None and key in settled:
+                return False
+            if states.get(key) == emoji:
+                if terminal_state:
+                    self._settle_reaction_key(key)
+                if not retain_state or terminal_state:
+                    states.pop(key, None)
+                return True
+            try:
+                if not self._bot:
+                    return False
+                await self._bot.set_message_reaction(
+                    chat_id=normalize_telegram_chat_id(chat_id),
+                    message_id=int(message_id),
+                    reaction=emoji,
+                )
+                if retain_state:
+                    states[key] = emoji
+                return True
+            except Exception as e:
+                logger.debug("[%s] set_message_reaction failed (%s): %s", self.name, emoji, _redact_telegram_error_text(e))
+                return False
+            finally:
+                if terminal_state:
+                    self._settle_reaction_key(key)
+                if not retain_state or terminal_state:
+                    states.pop(key, None)
+
+    async def _clear_reactions(
+        self, chat_id: str, message_id: str, *, terminal_state: bool = False
+    ) -> bool:
         """Clear all reactions from a Telegram message.
 
         Calling ``set_message_reaction`` with ``reaction=None`` (or an empty
         sequence) is the documented Bot API way to remove all bot-set
         reactions on a message — equivalent to Bot API 10.0's
         ``deleteMessageReaction`` but supported in PTB 22.6 already.
         """
-        if not self._bot:
-            return False
-        try:
-            await self._bot.set_message_reaction(
-                chat_id=normalize_telegram_chat_id(chat_id),
-                message_id=int(message_id),
-                reaction=None,
-            )
-            return True
-        except Exception as e:
-            logger.debug("[%s] clear reactions failed: %s", self.name, _redact_telegram_error_text(e))
-            return False
+        key = (str(chat_id), str(message_id))
+        async with self._reaction_update_lock():
+            try:
+                if not self._bot:
+                    return False
+                await self._bot.set_message_reaction(
+                    chat_id=normalize_telegram_chat_id(chat_id),
+                    message_id=int(message_id),
+                    reaction=None,
+                )
+                return True
+            except Exception as e:
+                logger.debug("[%s] clear reactions failed: %s", self.name, _redact_telegram_error_text(e))
+                return False
+            finally:
+                states = getattr(self, "_processing_reactions", None)
+                if states is not None:
+                    states.pop(key, None)
+                if terminal_state:
+                    self._settle_reaction_key(key)
+
+    async def on_processing_activity(
+        self,
+        source: "SessionSource",
+        message_id: str,
+        event_type: str,
+        tool_name: str | None = None,
+    ) -> None:
+        """Reflect thinking/tool state with Telegram-supported reactions."""
+        chat_id = source.chat_id
+        if not self._reactions_enabled() or not (chat_id and message_id):
+            return
+        if event_type in {"reasoning.available", "_thinking", "tool.completed", "tool.failed"}:
+            emoji = "🤔"
+        elif event_type == "tool.started":
+            name = (tool_name or "").lower()
+            if any(part in name for part in ("terminal", "browser", "computer", "code", "process")):
+                emoji = "👨‍💻"
+            elif any(part in name for part in ("file", "patch", "todo", "memory")):
+                emoji = "✍"
+            elif any(part in name for part in ("web", "search", "vision")):
+                emoji = "🤓"
+            elif any(part in name for part in ("delegate", "advisor", "clarify", "message")):
+                emoji = "🤝"
+            else:
+                emoji = "⚡"
+        else:
+            return
+        await self._set_reaction(chat_id, message_id, emoji, activity_state=True)
 
     async def on_processing_start(self, event: MessageEvent) -> None:
         """Add an in-progress reaction when message processing begins."""
@@ -9970,7 +10059,12 @@ async def on_processing_start(self, event: MessageEvent) -> None:
         chat_id = getattr(event.source, "chat_id", None)
         message_id = getattr(event, "message_id", None)
         if chat_id and message_id:
-            await self._set_reaction(chat_id, message_id, "\U0001f440")
+            await self._set_reaction(
+                chat_id,
+                message_id,
+                "\U0001f440",
+                begin_state=True,
+            )
 
     async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None:
         """Swap the in-progress reaction for a final success/failure reaction.
@@ -9992,12 +10086,14 @@ async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingO
         if not (chat_id and message_id):
             return
         if outcome == ProcessingOutcome.CANCELLED:
-            await self._clear_reactions(chat_id, message_id)
+            await self._clear_reactions(chat_id, message_id, terminal_state=True)
         else:
             await self._set_reaction(
                 chat_id,
                 message_id,
                 "\U0001f44d" if outcome == ProcessingOutcome.SUCCESS else "\U0001f44e",
+                retain_state=False,
+                terminal_state=True,
             )
 
 

tests/gateway/test_run_progress_topics.py

@@ -90,6 +90,17 @@ async def send_multiple_images(
         )
 
 
+class ActivityCaptureAdapter(ProgressCaptureAdapter):
+    def __init__(self, platform=Platform.TELEGRAM):
+        super().__init__(platform=platform)
+        self.activities = []
+
+    async def on_processing_activity(
+        self, source, message_id, event_type, tool_name=None
+    ) -> None:
+        self.activities.append((message_id, event_type, tool_name))
+
+
 class SmallLimitProgressAdapter(ProgressCaptureAdapter):
     """Adapter with a tiny platform limit to exercise progress rollover."""
 
@@ -401,6 +412,60 @@ def _make_runner(adapter):
     return runner
 
 
+@pytest.mark.asyncio
+async def test_activity_hook_receives_tool_events_when_visible_progress_is_off(
+    monkeypatch, tmp_path
+):
+    """Silent lifecycle status still receives tool events without progress bubbles."""
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = FakeAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+    (tmp_path / "config.yaml").write_text(
+        "display:\n"
+        "  platforms:\n"
+        "    telegram:\n"
+        "      tool_progress: false\n"
+        "      thinking_progress: false\n",
+        encoding="utf-8",
+    )
+
+    adapter = ActivityCaptureAdapter()
+    runner = _make_runner(adapter)
+    gateway_run = importlib.import_module("gateway.run")
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="123",
+        chat_type="private",
+    )
+
+    result = await runner._run_agent(
+        message="hello",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-activity-only",
+        session_key="agent:main:telegram:private:123",
+        event_message_id="incoming-42",
+    )
+    await asyncio.sleep(0.1)
+
+    assert result["final_response"] == "done"
+    assert adapter.sent == []
+    assert adapter.activities == [
+        ("incoming-42", "tool.started", "terminal"),
+        ("incoming-42", "tool.started", "browser_navigate"),
+    ]
+
+
 @pytest.mark.asyncio
 async def test_run_agent_progress_uses_event_message_id_for_slack_dm(monkeypatch, tmp_path):
     """Slack DM progress should keep event ts fallback threading."""