diff --git a/cli-config.yaml.example b/cli-config.yaml.example
index 963268d4ba68..554bc3a0268f 100644
--- a/cli-config.yaml.example
+++ b/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.
diff --git a/gateway/display_config.py b/gateway/display_config.py
index 832f5cb2f254..0103703422ef 100644
--- a/gateway/display_config.py
+++ b/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)
diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 5c2bbf96aa88..a40d61dab39b 100644
--- a/gateway/platforms/base.py
+++ b/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,13 +2141,37 @@ 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,
@@ -2110,22 +2179,44 @@ def pop_post_delivery_callback(
         *,
         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()
diff --git a/gateway/run.py b/gateway/run.py
index 15ce3ab08ce0..9f01e9047826 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -12836,6 +12836,27 @@ def _run_still_current() -> bool:
         
         # Queue for progress messages (thread-safe)
         progress_queue = queue.Queue() if tool_progress_enabled else None
+        temporary_tool_progress_enabled = (
+            source.platform == Platform.TELEGRAM
+            and bool(resolve_display_setting(
+                user_config,
+                platform_key,
+                "temporary_tool_progress",
+                False,
+            ))
+        )
+        progress_message_ids: list[str] = []
+        progress_cleanup_allowed = [False]
+        progress_cleanup_registered = [False]
+
+        def _track_progress_message_id(message_id) -> None:
+            """Track temporary Telegram progress/status bubbles for post-delivery cleanup."""
+            if not temporary_tool_progress_enabled or not message_id:
+                return
+            msg_id = str(message_id)
+            if msg_id not in progress_message_ids:
+                progress_message_ids.append(msg_id)
+
         last_tool = [None]  # Mutable container for tracking in closure
         last_progress_msg = [None]  # Track last message for dedup
         repeat_count = [0]  # How many times the same message repeated
@@ -13087,12 +13108,15 @@ async def send_progress_messages():
                                     adapter.name,
                                 )
                             can_edit = False
-                            await adapter.send(
+                            result = await adapter.send(
                                 chat_id=source.chat_id,
                                 content=msg,
                                 reply_to=_progress_reply_to,
                                 metadata=_progress_metadata,
                             )
+                            if result.success and result.message_id:
+                                progress_msg_id = result.message_id
+                                _track_progress_message_id(progress_msg_id)
                     else:
                         if can_edit:
                             # First tool: send all accumulated text as new message
@@ -13113,6 +13137,7 @@ async def send_progress_messages():
                             )
                         if result.success and result.message_id:
                             progress_msg_id = result.message_id
+                            _track_progress_message_id(progress_msg_id)
 
                     _last_edit_ts = time.monotonic()
 
@@ -13226,7 +13251,7 @@ def _status_callback_sync(event_type: str, message: str) -> None:
             if not _status_adapter or not _run_still_current():
                 return
             try:
-                asyncio.run_coroutine_threadsafe(
+                future = asyncio.run_coroutine_threadsafe(
                     _status_adapter.send(
                         _status_chat_id,
                         message,
@@ -13234,9 +13259,69 @@ def _status_callback_sync(event_type: str, message: str) -> None:
                     ),
                     _loop_for_step,
                 )
+
+                def _track_status_send_result(done_future) -> None:
+                    try:
+                        result = done_future.result()
+                        if getattr(result, "success", False):
+                            _track_progress_message_id(getattr(result, "message_id", None))
+                    except Exception as _track_err:
+                        logger.debug("status_callback tracking error (%s): %s", event_type, _track_err)
+
+                future.add_done_callback(_track_status_send_result)
             except Exception as _e:
                 logger.debug("status_callback error (%s): %s", event_type, _e)
 
+        def _register_tool_progress_cleanup_callback() -> None:
+            """Register best-effort cleanup for temporary Telegram progress bubbles."""
+            if progress_cleanup_registered[0] or not temporary_tool_progress_enabled:
+                return
+            adapter = self.adapters.get(source.platform)
+            if not adapter or not callable(getattr(adapter, "delete_message", None)):
+                return
+
+            async def _cleanup_progress_messages() -> None:
+                if not progress_cleanup_allowed[0]:
+                    return
+                ids_to_delete = list(dict.fromkeys(progress_message_ids))
+                for message_id in ids_to_delete:
+                    try:
+                        await adapter.delete_message(source.chat_id, message_id)
+                    except Exception as _cleanup_err:
+                        logger.debug(
+                            "Tool-progress cleanup failed (%s): %s",
+                            message_id,
+                            _cleanup_err,
+                        )
+
+            def _schedule_cleanup() -> None:
+                if not progress_cleanup_allowed[0] or not progress_message_ids:
+                    return
+                try:
+                    loop = asyncio.get_running_loop()
+                    loop.create_task(_cleanup_progress_messages())
+                except RuntimeError:
+                    asyncio.run_coroutine_threadsafe(
+                        _cleanup_progress_messages(),
+                        _loop_for_step,
+                    )
+
+            if getattr(type(adapter), "register_post_delivery_callback", None) is not None:
+                adapter.register_post_delivery_callback(
+                    session_key,
+                    _schedule_cleanup,
+                    generation=run_generation,
+                )
+            else:
+                _pdc = getattr(adapter, "_post_delivery_callbacks", None)
+                if isinstance(_pdc, dict):
+                    _pdc[session_key] = _schedule_cleanup
+                else:
+                    return
+            progress_cleanup_registered[0] = True
+
+        _register_tool_progress_cleanup_callback()
+
         def run_sync():
             # The conditional re-assignment of `message` further below
             # (prepending model-switch notes) makes Python treat it as a
@@ -14094,11 +14179,13 @@ async def _notify_long_running():
                     except Exception:
                         pass
                 try:
-                    await _notify_adapter.send(
+                    result = await _notify_adapter.send(
                         source.chat_id,
                         f"⏳ Still working... ({_elapsed_mins} min elapsed{_status_detail})",
                         metadata=_status_thread_metadata,
                     )
+                    if getattr(result, "success", False):
+                        _track_progress_message_id(getattr(result, "message_id", None))
                 except Exception as _ne:
                     logger.debug("Long-running notification error: %s", _ne)
 
@@ -14416,29 +14503,40 @@ async def _notify_long_running():
                         or _previewed
                     )
                     first_response = result.get("final_response", "")
+                    first_response_delivered = False
                     if first_response and not _already_streamed:
                         try:
                             logger.info(
                                 "Queued follow-up for session %s: final stream delivery not confirmed; sending first response before continuing.",
                                 session_key or "?",
                             )
-                            await adapter.send(
+                            send_result = await adapter.send(
                                 source.chat_id,
                                 first_response,
                                 metadata=_status_thread_metadata,
                             )
+                            first_response_delivered = bool(getattr(send_result, "success", False))
                         except Exception as e:
                             logger.warning("Failed to send first response before queued message: %s", e)
                     elif first_response:
+                        first_response_delivered = True
                         logger.info(
                             "Queued follow-up for session %s: skipping resend because final streamed delivery was confirmed.",
                             session_key or "?",
                         )
-                    # Release deferred bg-review notifications now that the
+                    if first_response_delivered:
+                        if progress_task and not progress_task.done():
+                            progress_task.cancel()
+                            try:
+                                await progress_task
+                            except asyncio.CancelledError:
+                                pass
+                        progress_cleanup_allowed[0] = True
+                    # Release deferred post-delivery callbacks now that the
                     # first response has been delivered.  Pop from the
                     # adapter's callback dict (prevents double-fire in
                     # base.py's finally block) and call it.
-                    if getattr(type(adapter), "pop_post_delivery_callback", None) is not None:
+                    if first_response_delivered and getattr(type(adapter), "pop_post_delivery_callback", None) is not None:
                         _bg_cb = adapter.pop_post_delivery_callback(
                             session_key,
                             generation=run_generation,
@@ -14448,7 +14546,7 @@ async def _notify_long_running():
                                 _bg_cb()
                             except Exception:
                                 pass
-                    elif adapter and hasattr(adapter, "_post_delivery_callbacks"):
+                    elif first_response_delivered and adapter and hasattr(adapter, "_post_delivery_callbacks"):
                         _bg_cb = adapter._post_delivery_callbacks.pop(session_key, None)
                         if callable(_bg_cb):
                             try:
@@ -14541,6 +14639,9 @@ async def _notify_long_running():
                     except asyncio.CancelledError:
                         pass
 
+        if isinstance(response, dict) and not response.get("failed"):
+            progress_cleanup_allowed[0] = True
+
         # If streaming already delivered the response, mark it so the
         # caller's send() is skipped (avoiding duplicate messages).
         # BUT: never suppress delivery when the agent failed — the error
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index cf2b0b528a6a..a89c83ac725e 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -841,7 +841,8 @@ def _ensure_hermes_home_managed(home: Path):
         # responses and content messages are never touched.  Default 0
         # (disabled) preserves prior behavior.
         "ephemeral_system_ttl": 0,
-        "platforms": {},  # Per-platform display overrides: {"telegram": {"tool_progress": "all"}, "slack": {"tool_progress": "off"}}
+        "temporary_tool_progress": False,  # Gateway: make progress bubbles temporary where supported (Telegram)
+        "platforms": {},  # Per-platform display overrides: {"telegram": {"tool_progress": "all", "temporary_tool_progress": True}, "slack": {"tool_progress": "off"}}
         # Gateway runtime-metadata footer appended to the FINAL message of a turn
         # (disabled by default to keep replies minimal). When enabled, renders
         # e.g. `model · 68% · ~/projects/hermes`. Per-platform overrides go under
diff --git a/tests/gateway/test_display_config.py b/tests/gateway/test_display_config.py
index 07d5c82a5f83..ca6847064bbd 100644
--- a/tests/gateway/test_display_config.py
+++ b/tests/gateway/test_display_config.py
@@ -78,6 +78,32 @@ def test_platform_override_only_affects_that_platform(self):
         assert resolve_display_setting(config, "slack", "tool_progress") == "off"
         assert resolve_display_setting(config, "telegram", "tool_progress") == "all"
 
+    def test_temporary_tool_progress_defaults_false(self):
+        from gateway.display_config import resolve_display_setting
+
+        assert resolve_display_setting({}, "telegram", "temporary_tool_progress") is False
+
+    def test_temporary_tool_progress_platform_override(self):
+        from gateway.display_config import resolve_display_setting
+
+        config = {
+            "display": {
+                "temporary_tool_progress": False,
+                "platforms": {"telegram": {"temporary_tool_progress": "true"}},
+            }
+        }
+        assert resolve_display_setting(config, "telegram", "temporary_tool_progress") is True
+
+    def test_temporary_tool_progress_reads_cleanup_alias(self):
+        from gateway.display_config import resolve_display_setting
+
+        config = {
+            "display": {
+                "platforms": {"telegram": {"cleanup_tool_progress": True}},
+            }
+        }
+        assert resolve_display_setting(config, "telegram", "temporary_tool_progress") is True
+
 
 # ---------------------------------------------------------------------------
 # Backward compatibility: tool_progress_overrides
diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py
index fb52e1e5863d..f90fff891017 100644
--- a/tests/gateway/test_run_progress_topics.py
+++ b/tests/gateway/test_run_progress_topics.py
@@ -65,6 +65,40 @@ async def edit_message(self, chat_id, message_id, content) -> SendResult:
         raise AssertionError("non-editable adapters should not receive edit_message calls")
 
 
+class FailingSendAdapter(ProgressCaptureAdapter):
+    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
+        self.sent.append(
+            {
+                "chat_id": chat_id,
+                "content": content,
+                "reply_to": reply_to,
+                "metadata": metadata,
+            }
+        )
+        return SendResult(success=False, error="synthetic send failure")
+
+
+class DeletingProgressCaptureAdapter(ProgressCaptureAdapter):
+    def __init__(self, platform=Platform.TELEGRAM):
+        super().__init__(platform=platform)
+        self.deleted = []
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
+        self.sent.append(
+            {
+                "chat_id": chat_id,
+                "content": content,
+                "reply_to": reply_to,
+                "metadata": metadata,
+            }
+        )
+        return SendResult(success=True, message_id=f"msg-{len(self.sent)}")
+
+    async def delete_message(self, chat_id, message_id) -> bool:
+        self.deleted.append({"chat_id": chat_id, "message_id": message_id})
+        return True
+
+
 class FakeAgent:
     def __init__(self, **kwargs):
         # Capture anything passed via kwargs (older code path) but don't
@@ -88,6 +122,69 @@ def run_conversation(self, message, conversation_history=None, task_id=None):
         }
 
 
+class FailingAfterProgressAgent(FakeAgent):
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        cb = self.tool_progress_callback
+        if cb is not None:
+            cb("tool.started", "terminal", "pwd", {})
+            time.sleep(0.35)
+        return {
+            "final_response": "",
+            "messages": [],
+            "api_calls": 1,
+            "failed": True,
+            "error": "synthetic failure",
+        }
+
+
+class LongRunningStatusAgent:
+    def __init__(self, **kwargs):
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        time.sleep(0.16)
+        return {
+            "final_response": "done",
+            "messages": [],
+            "api_calls": 1,
+        }
+
+
+class StatusCallbackAgent:
+    def __init__(self, **kwargs):
+        self.status_callback = kwargs.get("status_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        if self.status_callback:
+            self.status_callback("context_pressure", "⚠️ Compacting context...")
+            time.sleep(0.05)
+        return {
+            "final_response": "done",
+            "messages": [],
+            "api_calls": 1,
+        }
+
+
+class ProgressAndBackgroundReviewAgent(FakeAgent):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.background_review_callback = kwargs.get("background_review_callback")
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        cb = self.tool_progress_callback
+        if cb is not None:
+            cb("tool.started", "terminal", "pwd", {})
+            time.sleep(0.35)
+        if self.background_review_callback:
+            self.background_review_callback("💾 Skill 'progress-cleanup' created.")
+        return {
+            "final_response": "done",
+            "messages": [],
+            "api_calls": 1,
+        }
+
+
 class LongPreviewAgent:
     """Agent that emits a tool call with a very long preview string."""
     LONG_CMD = "cd /home/teknium/.hermes/hermes-agent/.worktrees/hermes-d8860339 && source .venv/bin/activate && python -m pytest tests/gateway/test_run_progress_topics.py -n0 -q"
@@ -568,6 +665,7 @@ async def _run_with_agent(
     chat_type="group",
     thread_id="17585",
     adapter_cls=ProgressCaptureAdapter,
+    run_generation=1,
 ):
     if config_data:
         import yaml
@@ -605,6 +703,8 @@ async def _run_with_agent(
             source=source,
             message_id="queued-1",
         )
+    if run_generation is not None:
+        runner._session_run_generation[session_key] = run_generation
 
     result = await runner._run_agent(
         message="hello",
@@ -613,6 +713,7 @@ async def _run_with_agent(
         source=source,
         session_id=session_id,
         session_key=session_key,
+        run_generation=run_generation,
     )
     return adapter, result
 
@@ -868,6 +969,103 @@ def _post_delivery_cb():
     assert released == [True]
 
 
+@pytest.mark.asyncio
+async def test_base_post_delivery_uses_generation_bound_during_handler():
+    adapter = ProgressCaptureAdapter()
+    session_key = "agent:main:telegram:dm:late-gen"
+    released = []
+
+    async def _handler(event):
+        adapter._active_sessions[session_key]._hermes_run_generation = 42
+        adapter.register_post_delivery_callback(
+            session_key,
+            lambda: released.append("current"),
+            generation=42,
+        )
+        return "done"
+
+    adapter.set_message_handler(_handler)
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="late-gen",
+        chat_type="dm",
+        thread_id=None,
+    )
+    event = MessageEvent(
+        text="hello",
+        message_type=MessageType.TEXT,
+        source=source,
+        message_id="msg-1",
+    )
+    adapter._active_sessions[session_key] = asyncio.Event()
+
+    await adapter._process_message_background(event, session_key)
+
+    assert released == ["current"]
+    assert adapter.pop_post_delivery_callback(session_key, generation=42) is None
+
+
+@pytest.mark.asyncio
+async def test_base_post_delivery_callbacks_are_composed_and_generation_safe():
+    adapter = ProgressCaptureAdapter()
+    calls = []
+
+    adapter.register_post_delivery_callback("sk", lambda: calls.append("old"), generation=1)
+    adapter.register_post_delivery_callback("sk", lambda: calls.append("first"), generation=2)
+    adapter.register_post_delivery_callback("sk", lambda: (_ for _ in ()).throw(RuntimeError("boom")), generation=2)
+    adapter.register_post_delivery_callback("sk", lambda: calls.append("second"), generation=2)
+    adapter.register_post_delivery_callback("sk", lambda: calls.append("late-stale"), generation=1)
+
+    callback = adapter.pop_post_delivery_callback("sk", generation=2)
+    assert callable(callback)
+    callback()
+
+    assert calls == ["first", "second"]
+
+    legacy_callback = adapter.pop_post_delivery_callback("sk")
+    assert legacy_callback is None
+
+    stale_callback = adapter.pop_post_delivery_callback("sk", generation=1)
+    assert stale_callback is None
+    assert calls == ["first", "second"]
+
+
+@pytest.mark.asyncio
+async def test_base_processing_does_not_release_post_delivery_callback_after_send_failure():
+    adapter = FailingSendAdapter()
+
+    async def _handler(event):
+        return "done"
+
+    adapter.set_message_handler(_handler)
+    released = []
+    adapter.register_post_delivery_callback("agent:main:telegram:dm:fail", lambda: released.append(True), generation=7)
+
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="fail",
+        chat_type="dm",
+        thread_id=None,
+    )
+    event = MessageEvent(
+        text="hello",
+        message_type=MessageType.TEXT,
+        source=source,
+        message_id="msg-1",
+    )
+    session_key = "agent:main:telegram:dm:fail"
+    active = asyncio.Event()
+    active._hermes_run_generation = 7
+    adapter._active_sessions[session_key] = active
+
+    await adapter._process_message_background(event, session_key)
+
+    assert adapter.sent and adapter.sent[0]["content"] == "done"
+    assert released == []
+    callback = adapter.pop_post_delivery_callback(session_key, generation=7)
+    assert callback is None
+
+
 @pytest.mark.asyncio
 async def test_run_agent_drops_tool_progress_after_generation_invalidation(monkeypatch, tmp_path):
     import yaml
@@ -1056,3 +1254,210 @@ async def test_verbose_mode_respects_explicit_tool_preview_length(monkeypatch, t
     assert VerboseAgent.LONG_CODE not in all_content
     # But should still contain the truncated portion with "..."
     assert "..." in all_content
+
+
+async def _fire_post_delivery_callback(adapter, session_key, generation=1):
+    callback = adapter.pop_post_delivery_callback(session_key, generation=generation)
+    assert callable(callback)
+    callback()
+    await asyncio.sleep(0.1)
+
+
+@pytest.mark.asyncio
+async def test_telegram_cleanup_tool_progress_deletes_after_post_delivery_callback(monkeypatch, tmp_path):
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        FakeAgent,
+        session_id="sess-cleanup-progress",
+        config_data={
+            "display": {
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                }
+            }
+        },
+        chat_id="12345",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+    session_key = "agent:main:telegram:dm:12345"
+
+    assert result["final_response"] == "done"
+    assert adapter.sent
+    assert adapter.deleted == []
+
+    await _fire_post_delivery_callback(adapter, session_key)
+
+    assert adapter.deleted == [{"chat_id": "12345", "message_id": "msg-1"}]
+
+
+@pytest.mark.asyncio
+async def test_telegram_temporary_tool_progress_deletes_still_working_notifications(monkeypatch, tmp_path):
+    monkeypatch.setenv("HERMES_AGENT_NOTIFY_INTERVAL", "0.05")
+
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        LongRunningStatusAgent,
+        session_id="sess-cleanup-still-working",
+        config_data={
+            "display": {
+                "tool_progress": "off",
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                },
+            }
+        },
+        chat_id="12349",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+    session_key = "agent:main:telegram:dm:12349"
+
+    assert result["final_response"] == "done"
+    still_working_ids = [
+        f"msg-{idx + 1}"
+        for idx, item in enumerate(adapter.sent)
+        if item["content"].startswith("⏳ Still working...")
+    ]
+    assert still_working_ids
+    assert adapter.deleted == []
+
+    await _fire_post_delivery_callback(adapter, session_key)
+
+    assert [item["message_id"] for item in adapter.deleted] == still_working_ids
+
+
+@pytest.mark.asyncio
+async def test_telegram_temporary_tool_progress_deletes_status_callback_messages(monkeypatch, tmp_path):
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        StatusCallbackAgent,
+        session_id="sess-cleanup-status-callback",
+        config_data={
+            "display": {
+                "tool_progress": "off",
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                },
+            }
+        },
+        chat_id="12350",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+    session_key = "agent:main:telegram:dm:12350"
+    await asyncio.sleep(0.1)
+
+    assert result["final_response"] == "done"
+    assert any(item["content"] == "⚠️ Compacting context..." for item in adapter.sent)
+    assert adapter.deleted == []
+
+    await _fire_post_delivery_callback(adapter, session_key)
+
+    assert adapter.deleted == [{"chat_id": "12350", "message_id": "msg-1"}]
+
+
+@pytest.mark.asyncio
+async def test_telegram_temporary_tool_progress_composes_with_background_review(monkeypatch, tmp_path):
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        ProgressAndBackgroundReviewAgent,
+        session_id="sess-cleanup-progress-bg-review",
+        config_data={
+            "display": {
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                }
+            }
+        },
+        chat_id="12347",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+    session_key = "agent:main:telegram:dm:12347"
+
+    assert result["final_response"] == "done"
+    assert adapter.deleted == []
+    assert not any("progress-cleanup" in item["content"] for item in adapter.sent)
+
+    await _fire_post_delivery_callback(adapter, session_key)
+
+    assert adapter.deleted == [{"chat_id": "12347", "message_id": "msg-1"}]
+    assert any("progress-cleanup" in item["content"] for item in adapter.sent)
+
+
+@pytest.mark.asyncio
+async def test_telegram_temporary_tool_progress_cleans_before_queued_followup(monkeypatch, tmp_path):
+    FakeAgent.calls = 0
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        FakeAgent,
+        session_id="sess-cleanup-progress-queued",
+        pending_text="queued follow-up",
+        config_data={
+            "display": {
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                }
+            }
+        },
+        chat_id="12348",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+
+    assert result["final_response"] == "done"
+    await asyncio.sleep(0.1)
+
+    assert {item["message_id"] for item in adapter.deleted} >= {"msg-1"}
+
+
+@pytest.mark.asyncio
+async def test_telegram_cleanup_tool_progress_keeps_breadcrumbs_when_agent_failed(monkeypatch, tmp_path):
+    adapter, result = await _run_with_agent(
+        monkeypatch,
+        tmp_path,
+        FailingAfterProgressAgent,
+        session_id="sess-cleanup-progress-failure",
+        config_data={
+            "display": {
+                "platforms": {
+                    "telegram": {
+                        "temporary_tool_progress": True,
+                    }
+                }
+            }
+        },
+        chat_id="12346",
+        chat_type="dm",
+        thread_id=None,
+        adapter_cls=DeletingProgressCaptureAdapter,
+    )
+    session_key = "agent:main:telegram:dm:12346"
+
+    assert result["failed"] is True
+    assert adapter.sent
+
+    await _fire_post_delivery_callback(adapter, session_key)
+
+    assert adapter.deleted == []