fix(compression): route the heartbeat through the shared compaction status contract

The contributor heartbeat called status_callback("compacting", <ad-hoc text>).
Every other compaction status uses the "lifecycle" key: the TUI gateway
re-tags lifecycle statuses to kind="compacting" via
is_compaction_progress_status, Telegram edits one bubble per status key, and
gateway/run.py's chat-platform filter only recognises the registered
templates — so the ad-hoc text leaked to Telegram/Discord on every tick with
compression.progress_notices off (verified: _prepare_gateway_status_message
passed it through). Register COMPACTION_HEARTBEAT_STATUS next to
COMPACTION_STATUS, emit it under "lifecycle", and cover the filter.

agent/conversation_compression.py

@@ -109,6 +109,12 @@ COMPACTION_STATUS_MARKER = "Compacting context"
 COMPACTION_STATUS = (
     f"🗜️ {COMPACTION_STATUS_MARKER} — summarizing earlier conversation so I can continue..."
 )
+# Periodic heartbeat re-emitted while a long compression is still running so
+# remote transports with idle-turn watchdogs (#98371) see progress. Same
+# marker as COMPACTION_STATUS so every consumer classifies it identically.
+COMPACTION_HEARTBEAT_STATUS = (
+    f"🗜️ {COMPACTION_STATUS_MARKER} — still summarizing earlier conversation so I can continue..."
+)
 
 COMPACTION_DONE_STATUS = "✓ Context compaction complete — continuing turn..."
 
@@ -2414,11 +2420,12 @@ class _CompressionActivityHeartbeat:
         if not status_callback:
             return
         try:
-            status_callback(
-                "compacting",
-                f"🗜️ {COMPACTION_STATUS_MARKER} — still summarizing "
-                "earlier conversation so I can continue...",
-            )
+            # Same "lifecycle" key as every other compaction status: the
+            # TUI gateway re-tags it to kind="compacting" via
+            # is_compaction_progress_status, and Telegram's per-key
+            # send_or_update_status edits the existing bubble instead of
+            # appending one per tick.
+            status_callback("lifecycle", COMPACTION_HEARTBEAT_STATUS)
         except Exception:
             logger.debug(
                 "status_callback error in compression heartbeat", exc_info=True
@@ -6151,6 +6158,7 @@ def try_shrink_image_parts_in_messages(
 __all__ = [
     "COMPACTION_STATUS",
     "COMPACTION_DONE_STATUS",
+    "COMPACTION_HEARTBEAT_STATUS",
     "COMPACTION_STATUS_MARKER",
     "is_compaction_progress_status",
     "check_compression_model_feasibility",

gateway/run.py

@@ -51,6 +51,7 @@ from typing import Awaitable, Callable, Dict, Optional, Any, List, Tuple, Union,
 from agent.async_utils import consume_detached_task_result, safe_schedule_threadsafe
 from agent.conversation_compression import (
     COMPACTION_DONE_STATUS,
+    COMPACTION_HEARTBEAT_STATUS,
     COMPACTION_STATUS,
     COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE,
     COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE,
@@ -131,7 +132,7 @@ _TELEGRAM_NOISY_STATUS_RE = re.compile(
     r"|auto-lowered\s+(?:this\s+)?session'?s?\s+threshold"
     r"|configured\s+auxiliary\s+compression\s+provider\s+.+\s+unavailable"
     r"|skipping\s+concurrent\s+compression"
-    r"|compacting\s+context\s+[—-]\s+summarizing\s+earlier\s+conversation"
+    r"|compacting\s+context\s+[—-]\s+(?:still\s+)?summarizing\s+earlier\s+conversation"
     r"|resumed\s+after\s+\d+s\s+idle\s+[—-]\s+compacting"
     r"|preflight\s+compression"
     r"|pre[- ]api\s+compression"
@@ -533,6 +534,7 @@ _COMPRESSION_PROGRESS_STATUS_RE = re.compile(
         _status_template_to_regex(_template)
         for _template in (
             COMPACTION_STATUS,
+            COMPACTION_HEARTBEAT_STATUS,
             COMPACTION_DONE_STATUS,
             PRE_API_COMPRESSION_STATUS_TEMPLATE,
             PREFLIGHT_COMPRESSION_STATUS_TEMPLATE,

tests/agent/test_compression_concurrent_fork.py

@@ -203,10 +203,23 @@ def test_compression_activity_heartbeat_emits_client_status_events(tmp_path: Pat
 
     agent._compress_context(messages, "sys", approx_tokens=120_000)
 
-    compacting = [event for event, _message in status_events if event == "compacting"]
+    from agent.conversation_compression import (
+        COMPACTION_HEARTBEAT_STATUS,
+        is_compaction_progress_status,
+    )
+
+    heartbeats = [
+        (event, message)
+        for event, message in status_events
+        if message == COMPACTION_HEARTBEAT_STATUS
+    ]
     # One emit at heartbeat start plus at least one periodic tick while the
     # summary call blocks.
-    assert len(compacting) >= 2
+    assert len(heartbeats) >= 2
+    # Same "lifecycle" key as the other compaction statuses so the TUI gateway
+    # re-tags it to kind="compacting" and Telegram edits one bubble in place.
+    assert {event for event, _ in heartbeats} == {"lifecycle"}
+    assert is_compaction_progress_status(COMPACTION_HEARTBEAT_STATUS)
 
 
 def test_lock_contender_preserves_terminal_compaction_lifecycle(tmp_path: Path) -> None:

tests/gateway/test_compaction_heartbeat_gateway_filter.py

@@ -0,0 +1,36 @@
+"""The compaction heartbeat (#98371) must be classified like COMPACTION_STATUS.
+
+Chat platforms suppress routine compression chatter unless
+``compression.progress_notices`` is enabled; a heartbeat that slipped past
+that gate would post a bubble per tick to Telegram/Discord. The TUI gateway,
+by contrast, must keep receiving it so idle-turn watchdogs see progress.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from agent.conversation_compression import (
+    COMPACTION_HEARTBEAT_STATUS,
+    COMPACTION_STATUS,
+    is_compaction_progress_status,
+)
+from gateway.run import _prepare_gateway_status_message
+
+
+def _telegram():
+    return SimpleNamespace(value="telegram")
+
+
+def test_heartbeat_is_compaction_progress_for_tui_retagging():
+    assert is_compaction_progress_status(COMPACTION_HEARTBEAT_STATUS)
+
+
+def test_heartbeat_suppressed_on_chat_platforms_by_default():
+    with patch("gateway.run._gateway_compression_progress_notices_enabled", return_value=False):
+        assert _prepare_gateway_status_message(_telegram(), "lifecycle", COMPACTION_STATUS) is None
+        assert _prepare_gateway_status_message(_telegram(), "lifecycle", COMPACTION_HEARTBEAT_STATUS) is None
+
+
+def test_heartbeat_passes_when_progress_notices_enabled():
+    with patch("gateway.run._gateway_compression_progress_notices_enabled", return_value=True):
+        assert _prepare_gateway_status_message(_telegram(), "lifecycle", COMPACTION_HEARTBEAT_STATUS) == COMPACTION_HEARTBEAT_STATUS