feat(gateway): add Telegram temporary tool progress

cli-config.yaml.example

@@ -875,6 +875,13 @@ display:
   # Toggle at runtime with /verbose in the CLI
   tool_progress: all
 
+  # Platform-specific display overrides.
+  # Example: show Telegram tool progress while work is active, then delete
+  # temporary bot-sent progress bubbles after the final response succeeds.
+  platforms:
+    telegram:
+      temporary_tool_progress: false
+
   # Gateway-only natural mid-turn assistant updates.
   # When true, completed assistant status messages are sent as separate chat
   # messages. This is independent of tool_progress and gateway streaming.

gateway/display_config.py

@@ -32,6 +32,7 @@
 
 _GLOBAL_DEFAULTS: dict[str, Any] = {
     "tool_progress": "all",
+    "temporary_tool_progress": False,
     "show_reasoning": False,
     "tool_preview_length": 0,
     "streaming": None,  # None = follow top-level streaming config
@@ -140,6 +141,12 @@ def resolve_display_setting(
         val = plat_overrides.get(setting)
         if val is not None:
             return _normalise(setting, val)
+        # Temporary compatibility for the local rollout name used before the
+        # upstream-facing setting was named by behavior rather than mechanism.
+        if setting == "temporary_tool_progress":
+            val = plat_overrides.get("cleanup_tool_progress")
+            if val is not None:
+                return _normalise(setting, val)
 
     # 1b. Backward compat: display.tool_progress_overrides.<platform>
     if setting == "tool_progress":
@@ -156,6 +163,10 @@ def resolve_display_setting(
         val = display_cfg.get(setting)
         if val is not None:
             return _normalise(setting, val)
+        if setting == "temporary_tool_progress":
+            val = display_cfg.get("cleanup_tool_progress")
+            if val is not None:
+                return _normalise(setting, val)
 
     # 3. Built-in platform default
     plat_defaults = _PLATFORM_DEFAULTS.get(platform_key)
@@ -184,7 +195,7 @@ def _normalise(setting: str, value: Any) -> Any:
         if value is True:
             return "all"
         return str(value).lower()
-    if setting in ("show_reasoning", "streaming"):
+    if setting in ("temporary_tool_progress", "show_reasoning", "streaming"):
         if isinstance(value, str):
             return value.lower() in ("true", "1", "yes", "on")
         return bool(value)

gateway/platforms/base.py

@@ -2085,6 +2085,51 @@ async def interrupt_session_activity(self, session_key: str, chat_id: str) -> No
         except Exception:
             pass
 
+    @staticmethod
+    def _normalise_post_delivery_callbacks(entry: Any) -> list[tuple[int | None, Callable]]:
+        """Return ``entry`` as a list of ``(generation, callback)`` pairs.
+
+        Older code stored a single callback or ``(generation, callback)`` tuple.
+        Newer code appends callbacks so independent post-delivery features do
+        not clobber each other.  Keep the reader tolerant so in-flight tests and
+        platform subclasses using the old shape continue to work.
+        """
+        if entry is None:
+            return []
+        if callable(entry):
+            return [(None, entry)]
+        if isinstance(entry, tuple) and len(entry) == 2 and callable(entry[1]):
+            gen = entry[0]
+            return [(int(gen) if gen is not None else None, entry[1])]
+        if isinstance(entry, list):
+            callbacks: list[tuple[int | None, Callable]] = []
+            for item in entry:
+                if callable(item):
+                    callbacks.append((None, item))
+                    continue
+                if isinstance(item, tuple) and len(item) == 2 and callable(item[1]):
+                    gen = item[0]
+                    callbacks.append((int(gen) if gen is not None else None, item[1]))
+            return callbacks
+        return []
+
+    @staticmethod
+    def _compose_post_delivery_callbacks(callbacks: list[Callable]) -> Callable | None:
+        """Compose callbacks with per-callback exception isolation."""
+        if not callbacks:
+            return None
+        if len(callbacks) == 1:
+            return callbacks[0]
+
+        def _combined_callback() -> None:
+            for callback in callbacks:
+                try:
+                    callback()
+                except Exception:
+                    logger.debug("Post-delivery callback failed", exc_info=True)
+
+        return _combined_callback
+
     def register_post_delivery_callback(
         self,
         session_key: str,
@@ -2096,36 +2141,82 @@ def register_post_delivery_callback(
 
         ``generation`` lets callers tie the callback to a specific gateway run
         generation so stale runs cannot clear callbacks owned by a fresher run.
+        Multiple callbacks for the same session/generation are composed instead
+        of replacing each other.
         """
         if not session_key or not callable(callback):
             return
-        if generation is None:
-            self._post_delivery_callbacks[session_key] = callback
-        else:
-            self._post_delivery_callbacks[session_key] = (int(generation), callback)
+        normalized_generation = int(generation) if generation is not None else None
+        existing_callbacks = self._normalise_post_delivery_callbacks(
+            self._post_delivery_callbacks.get(session_key)
+        )
+        if normalized_generation is not None:
+            newer_generations = [
+                entry_generation
+                for entry_generation, _ in existing_callbacks
+                if entry_generation is not None and entry_generation > normalized_generation
+            ]
+            if newer_generations:
+                # A stale run is trying to register after a newer run already owns
+                # the session callbacks.  Preserve the newer callbacks rather than
+                # reintroducing the stale-generation clobber bug.
+                return
+        elif any(entry_generation is not None for entry_generation, _ in existing_callbacks):
+            # Legacy/unknown-generation callbacks should not clobber callbacks
+            # explicitly owned by a known gateway run generation.
+            return
+        callbacks = [
+            (entry_generation, cb)
+            for entry_generation, cb in existing_callbacks
+            if entry_generation == normalized_generation
+        ]
+        callbacks.append((normalized_generation, callback))
+        self._post_delivery_callbacks[session_key] = callbacks[0] if len(callbacks) == 1 else callbacks
 
     def pop_post_delivery_callback(
         self,
         session_key: str,
         *,
         generation: int | None = None,
     ) -> Callable | None:
-        """Pop a deferred callback, optionally requiring generation ownership."""
+        """Pop deferred callback(s), optionally requiring generation ownership."""
         if not session_key:
             return None
         entry = self._post_delivery_callbacks.get(session_key)
-        if entry is None:
+        callbacks = self._normalise_post_delivery_callbacks(entry)
+        if not callbacks:
             return None
-        if isinstance(entry, tuple) and len(entry) == 2:
-            entry_generation, callback = entry
-            if generation is not None and int(entry_generation) != int(generation):
-                return None
+
+        if generation is None:
+            # Unknown generation is legacy-only: do not pop generation-tagged
+            # callbacks, or an old task can clobber callbacks owned by a newer
+            # gateway run for the same session.
+            matched = [cb for entry_generation, cb in callbacks if entry_generation is None]
+            remaining = [
+                (entry_generation, cb)
+                for entry_generation, cb in callbacks
+                if entry_generation is not None
+            ]
+            if remaining:
+                self._post_delivery_callbacks[session_key] = remaining[0] if len(remaining) == 1 else remaining
+            else:
+                self._post_delivery_callbacks.pop(session_key, None)
+            return self._compose_post_delivery_callbacks(matched)
+
+        target_generation = int(generation)
+        matched: list[Callable] = []
+        remaining: list[tuple[int | None, Callable]] = []
+        for entry_generation, callback in callbacks:
+            if entry_generation == target_generation:
+                matched.append(callback)
+            else:
+                remaining.append((entry_generation, callback))
+
+        if remaining:
+            self._post_delivery_callbacks[session_key] = remaining[0] if len(remaining) == 1 else remaining
+        else:
             self._post_delivery_callbacks.pop(session_key, None)
-            return callback if callable(callback) else None
-        if generation is not None:
-            return None
-        self._post_delivery_callbacks.pop(session_key, None)
-        return entry if callable(entry) else None
+        return self._compose_post_delivery_callbacks(matched)
 
     # ── Processing lifecycle hooks ──────────────────────────────────────────
     # Subclasses override these to react to message processing events
@@ -3049,13 +3140,26 @@ async def _stop_typing_task() -> None:
                 "_hermes_run_generation",
                 None,
             )
-            if hasattr(self, "pop_post_delivery_callback"):
-                _post_cb = self.pop_post_delivery_callback(
-                    session_key,
-                    generation=_callback_generation,
-                )
+            _post_cb = None
+            if delivery_attempted and not delivery_succeeded:
+                # Final delivery failed: discard callbacks for this exact run
+                # without firing them.  Leaving them registered would let a
+                # later successful run delete/release stale artifacts.
+                if hasattr(self, "pop_post_delivery_callback"):
+                    self.pop_post_delivery_callback(
+                        session_key,
+                        generation=_callback_generation,
+                    )
+                elif _callback_generation is None:
+                    getattr(self, "_post_delivery_callbacks", {}).pop(session_key, None)
             else:
-                _post_cb = getattr(self, "_post_delivery_callbacks", {}).pop(session_key, None)
+                if hasattr(self, "pop_post_delivery_callback"):
+                    _post_cb = self.pop_post_delivery_callback(
+                        session_key,
+                        generation=_callback_generation,
+                    )
+                else:
+                    _post_cb = getattr(self, "_post_delivery_callbacks", {}).pop(session_key, None)
             if callable(_post_cb):
                 try:
                     _post_cb()