feat(signal): add timestamp edits and opt-in tool progress

gateway/display_config.py

@@ -150,7 +150,8 @@
     "matrix":          _TIER_MEDIUM,
     "feishu":          _TIER_MEDIUM,
 
-    # Tier 3 — no edit support, progress messages are permanent
+    # Tier 3 — low-noise defaults. Signal can edit timestamp-addressed
+    # messages, but keeps token streaming and tool progress off until opt-in.
     "signal":          _TIER_LOW,
     "whatsapp":        _TIER_MEDIUM,  # Baileys bridge supports /edit
     # WhatsApp Cloud API: Meta added message editing in 2023 but the

gateway/platforms/base.py

@@ -2484,6 +2484,28 @@ class SendResult:
     error_kind: Optional[str] = None
 
 
+def next_edit_target_message_id(
+    adapter: Any,
+    current_message_id: Optional[str],
+    result: Any,
+) -> Optional[str]:
+    """Return the platform handle that a subsequent edit must target.
+
+    Most adapters keep editing the original message id even when an edit API
+    returns a fresh replacement/event id (Matrix is the important example).
+    Timestamp-chained platforms opt in explicitly so shared callers never
+    infer this contract from ``SendResult.message_id`` alone.
+    """
+    if getattr(adapter, "EDIT_RESULT_ID_IS_NEXT_TARGET", False) is not True:
+        return current_message_id
+    if not (result and getattr(result, "success", False)):
+        return current_message_id
+    next_message_id = getattr(result, "message_id", None)
+    if next_message_id is None or next_message_id == "":
+        return current_message_id
+    return str(next_message_id)
+
+
 # Machine-readable send-failure categories.  Kept platform-neutral so every
 # adapter can populate ``SendResult.error_kind`` from the same vocabulary and
 # the gateway can decide — once, in one place — whether a failure is worth
@@ -2938,6 +2960,31 @@ def set_status_text(self, chat_id: str, text: Optional[str]) -> None:
     # set this to False to stay correct-by-default.
     supports_async_delivery: bool = True
 
+    # Whether this adapter supports explicit ``edit_message`` calls for
+    # previously sent messages.  Most chat platforms support this; non-editable
+    # adapters override to False so callers can avoid pretending a message can
+    # be mutated later.
+    SUPPORTS_MESSAGE_EDITING: bool = True
+
+    # Whether edit-based high-frequency response streaming is safe.  Defaults
+    # to SUPPORTS_MESSAGE_EDITING via gateway.run's helper; set explicitly when
+    # a platform can handle operator/user-requested edits but should not be
+    # driven by the stream consumer cadence.
+    SUPPORTS_STREAMING_EDITS: Optional[bool] = None
+
+    # Whether edit-based tool/thinking progress bubbles are safe.  Defaults to
+    # the streaming-edit capability via gateway.run's helper; adapters can set
+    # this separately if token streaming and progress bubbles have different
+    # platform costs.
+    SUPPORTS_PROGRESS_EDITS: Optional[bool] = None
+
+    # Whether a successful edit's SendResult.message_id becomes the required
+    # target for the NEXT edit. False by default: replacement-event APIs such
+    # as Matrix return a fresh event id while subsequent m.replace operations
+    # must continue targeting the original event. Signal timestamp chains set
+    # this True because each edit mints the next editTimestamp anchor.
+    EDIT_RESULT_ID_IS_NEXT_TARGET: bool = False
+
     # Whether this adapter's ``send()`` splits long content into multiple
     # messages via ``truncate_message()``.  When True, the delivery router
     # (gateway/delivery.py) skips gateway-level truncation and lets the

gateway/platforms/signal.py

@@ -254,10 +254,15 @@ class SignalAdapter(BasePlatformAdapter):
     """Signal messenger adapter using signal-cli HTTP daemon."""
 
     platform = Platform.SIGNAL
-    # Signal has no real edit API for already-sent messages. Mark it explicitly
-    # so streaming suppresses the visible cursor instead of leaving a stale tofu
-    # square behind in chat clients when edit attempts fail.
-    SUPPORTS_MESSAGE_EDITING = False
+    # signal-cli exposes message edits via send(editTimestamp=...). Keep
+    # token-by-token response streaming disabled, but allow the lower-frequency
+    # accumulated tool-progress bubble when the user explicitly enables it.
+    # Signal stays on the tier-low display default, so progress is off unless
+    # /verbose or a config override opts in.
+    SUPPORTS_MESSAGE_EDITING = True
+    SUPPORTS_STREAMING_EDITS = False
+    SUPPORTS_PROGRESS_EDITS = True
+    EDIT_RESULT_ID_IS_NEXT_TARGET = True
 
     def __init__(self, config: PlatformConfig):
         super().__init__(config, Platform.SIGNAL)
@@ -1050,22 +1055,22 @@ def _validate_send_result(self, result: Any) -> tuple[bool, Optional[str]]:
     # Sending
     # ------------------------------------------------------------------
 
-    async def send(
+    async def _build_send_params(
         self,
         chat_id: str,
         content: str,
-        reply_to: Optional[str] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-    ) -> SendResult:
-        """Send a text message with native Signal formatting."""
-        await self._stop_typing_indicator(chat_id)
-
+        *,
+        edit_timestamp: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """Build signal-cli JSON-RPC ``send`` params for text sends/edits."""
         plain_text, text_styles = self._markdown_to_signal(content)
 
         params: Dict[str, Any] = {
             "account": self.account,
             "message": plain_text,
         }
+        if edit_timestamp is not None:
+            params["editTimestamp"] = edit_timestamp
 
         if text_styles:
             if len(text_styles) == 1:
@@ -1078,20 +1083,96 @@ async def send(
         else:
             params["recipient"] = [await self._resolve_recipient(chat_id)]
 
-        logger.info("[Signal] Sending response (%d chars) to %s", len(plain_text), chat_id)
+        return params
+
+    @staticmethod
+    def _extract_send_timestamp(rpc_result: Any) -> Optional[str]:
+        """Return signal-cli's send timestamp as an editable message id."""
+        if isinstance(rpc_result, dict):
+            timestamp = rpc_result.get("timestamp")
+        else:
+            timestamp = None
+        if timestamp is None or timestamp == "":
+            return None
+        return str(timestamp)
+
+    async def send(
+        self,
+        chat_id: str,
+        content: str,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send a text message with native Signal formatting."""
+        await self._stop_typing_indicator(chat_id)
+
+        params = await self._build_send_params(chat_id, content)
+        logger.info("[Signal] Sending response (%d chars) to %s", len(params["message"]), chat_id)
         result = await self._rpc("send", params)
 
         if result is not None:
             success, err_msg = self._validate_send_result(result)
             if not success:
                 return SendResult(success=False, error=err_msg, raw_response=result)
             self._track_sent_timestamp(result)
-            # Signal has no editable message identifier. Returning None keeps the
-            # stream consumer on the non-edit fallback path instead of pretending
-            # future edits can remove an in-progress cursor from the chat thread.
-            return SendResult(success=True, message_id=None)
+            return SendResult(
+                success=True,
+                message_id=self._extract_send_timestamp(result),
+            )
         return SendResult(success=False, error="RPC send failed")
 
+    async def edit_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        content: str,
+        *,
+        finalize: bool = False,
+    ) -> SendResult:
+        """Edit a previously sent Signal message via signal-cli editTimestamp.
+
+        signal-cli's JSON-RPC API mirrors the CLI's ``send --edit-timestamp``
+        option: edits are sent through ``send`` with ``editTimestamp`` set to
+        the current Signal timestamp stored as Hermes' ``message_id``.
+        signal-cli returns a new timestamp for the edit event; propagate that
+        fresh id so a chain of later edits targets the newest edit handle.
+        """
+        del finalize  # Signal edits have no explicit streaming finalization.
+
+        if not message_id:
+            return SendResult(success=False, error="Signal edit requires message_id timestamp")
+
+        try:
+            edit_timestamp = int(str(message_id))
+        except (TypeError, ValueError):
+            return SendResult(
+                success=False,
+                error="Signal edit requires numeric message_id timestamp",
+            )
+
+        await self._stop_typing_indicator(chat_id)
+        params = await self._build_send_params(
+            chat_id,
+            content,
+            edit_timestamp=edit_timestamp,
+        )
+        result = await self._rpc("send", params)
+
+        if result is not None:
+            success, err_msg = self._validate_send_result(result)
+            if not success:
+                return SendResult(success=False, error=err_msg, raw_response=result)
+            fresh_timestamp = self._extract_send_timestamp(result)
+            if fresh_timestamp is None:
+                return SendResult(
+                    success=False,
+                    error="Signal edit response missing fresh timestamp",
+                    raw_response=result,
+                )
+            self._track_sent_timestamp(result)
+            return SendResult(success=True, message_id=fresh_timestamp)
+        return SendResult(success=False, error="RPC edit failed")
+
     def _track_sent_timestamp(self, rpc_result) -> None:
         """Record outbound message timestamp for echo-back filtering."""
         ts = rpc_result.get("timestamp") if isinstance(rpc_result, dict) else None

gateway/run.py

@@ -831,6 +831,34 @@ def _resolve_progress_thread_id(
     return None
 
 
+def _adapter_supports_streaming_edits(adapter: Any) -> bool:
+    """Return whether adapter edits are safe for high-frequency streaming.
+
+    SUPPORTS_MESSAGE_EDITING means explicit user/operator edits are possible.
+    Streaming is a narrower capability: some platforms expose an edit API but
+    make every edit a visible high-frequency event, so they should opt out of
+    response-stream editing while keeping explicit edit_message().
+    """
+    streaming_capability = getattr(adapter, "SUPPORTS_STREAMING_EDITS", None)
+    if streaming_capability is not None:
+        return bool(streaming_capability)
+    return bool(getattr(adapter, "SUPPORTS_MESSAGE_EDITING", True))
+
+
+def _adapter_supports_progress_edits(adapter: Any) -> bool:
+    """Return whether adapter edits are safe for tool/thinking progress.
+
+    Progress bubbles are throttled and lower-frequency than token streaming,
+    but they are still automatic edits. Platforms may expose explicit
+    edit_message() for deliberate user/operator edits while choosing a separate
+    policy for automatic progress edits.
+    """
+    progress_capability = getattr(adapter, "SUPPORTS_PROGRESS_EDITS", None)
+    if progress_capability is not None:
+        return bool(progress_capability)
+    return _adapter_supports_streaming_edits(adapter)
+
+
 def _has_platform_display_override(user_config: dict, platform_key: str, setting: str) -> bool:
     """Return True when display.platforms.<platform> explicitly sets setting."""
     display = user_config.get("display") if isinstance(user_config, dict) else None
@@ -2433,6 +2461,7 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
     _reply_anchor_for_event,
     build_auto_tts_output_path,
     merge_pending_message_event,
+    next_edit_target_message_id,
     utf16_len,
 )
 from gateway.shutdown_watchdog import (
@@ -4042,14 +4071,18 @@ async def send_progress_messages(self):
         if not adapter:
             return
 
-        # Skip tool progress for platforms that don't support message
-        # editing (e.g. iMessage/BlueBubbles) — each progress update
-        # would become a separate message bubble, which is noisy.
+        # Skip tool/thinking progress for platforms that cannot safely edit
+        # progress bubbles. An adapter may support explicit edits while
+        # independently opting out of automatic progress cadence.
         # getattr, not attribute access: duck-typed adapters (test fakes,
         # minimal plugin adapters) may not define edit_message at all —
         # "missing" means the same thing as "base no-op": can't edit.
         _adapter_edit = getattr(type(adapter), "edit_message", None)
-        if _adapter_edit is None or _adapter_edit is BasePlatformAdapter.edit_message:
+        if (
+            not _adapter_supports_progress_edits(adapter)
+            or _adapter_edit is None
+            or _adapter_edit is BasePlatformAdapter.edit_message
+        ):
             while not ctx.progress_queue.empty():
                 try:
                     ctx.progress_queue.get_nowait()
@@ -4107,6 +4140,7 @@ async def send_progress_messages(self):
                 _edit_accepts_metadata = False
 
         async def _edit_progress_message(message_id: str, content: str):
+            nonlocal progress_msg_id
             kwargs = {
                 "chat_id": ctx.source.chat_id,
                 "message_id": message_id,
@@ -4116,7 +4150,13 @@ async def _edit_progress_message(message_id: str, content: str):
                 kwargs["finalize"] = True
             if _edit_accepts_metadata:
                 kwargs["metadata"] = ctx._progress_metadata
-            return await adapter.edit_message(**kwargs)
+            result = await adapter.edit_message(**kwargs)
+            progress_msg_id = next_edit_target_message_id(
+                adapter,
+                message_id,
+                result,
+            )
+            return result
 
         def _progress_text(lines: list) -> str:
             return "\n".join(str(line) for line in lines)
@@ -4334,6 +4374,11 @@ async def _roll_progress_overflow_if_needed() -> bool:
                         progress_msg_id = result.message_id
                         if ctx._cleanup_progress:
                             ctx._cleanup_msg_ids.append(str(result.message_id))
+                    elif result.success and can_edit:
+                        # The message was delivered but cannot be addressed for
+                        # a later edit. Degrade to one new line per update
+                        # instead of replaying the accumulated transcript.
+                        can_edit = False
 
                 _last_edit_ts = time.monotonic()
 
@@ -24536,9 +24581,8 @@ def _build_stream_consumer_config(
         """Build the shared ``StreamConsumerConfig`` and the optional
         Telegram pause-typing closure used by both agent-run paths.
 
-        ``on_missing_cursor`` controls how platforms whose adapter sets
-        ``SUPPORTS_MESSAGE_EDITING = False`` are handled — both semantics
-        are preserved verbatim from the pre-refactor call sites:
+        ``on_missing_cursor`` controls how legacy platforms whose adapter sets
+        ``SUPPORTS_MESSAGE_EDITING = False`` are handled:
 
         - ``"fallback"`` (proxy path): stream anyway with an empty cursor.
         - ``"raise"`` (in-process agent path): raise ``RuntimeError`` so
@@ -24562,8 +24606,16 @@ def _pause_typing_before_finalize(
         # duplicate messages (partial + final).
         # (The proxy path instead opts into a cursorless fallback
         # via on_missing_cursor="fallback".)
-        _adapter_supports_edit = getattr(adapter, "SUPPORTS_MESSAGE_EDITING", True)
-        if not _adapter_supports_edit and on_missing_cursor == "raise":
+        _adapter_supports_edit = _adapter_supports_streaming_edits(adapter)
+        _explicit_streaming_capability = getattr(
+            adapter,
+            "SUPPORTS_STREAMING_EDITS",
+            None,
+        )
+        if not _adapter_supports_edit and (
+            on_missing_cursor == "raise"
+            or _explicit_streaming_capability is not None
+        ):
             raise RuntimeError("skip streaming for non-editable platform")
         _effective_cursor = scfg.cursor if _adapter_supports_edit else ""
         # Some Matrix clients render the streaming cursor
@@ -25821,7 +25873,13 @@ async def _notify_long_running():
                         except Exception as _ee:
                             logger.debug("Heartbeat edit failed: %s", _ee)
                             _notify_res = None
-                    if not (_notify_res and getattr(_notify_res, "success", False)):
+                    if _notify_res and getattr(_notify_res, "success", False):
+                        _heartbeat_msg_id = next_edit_target_message_id(
+                            _notify_adapter,
+                            _heartbeat_msg_id,
+                            _notify_res,
+                        )
+                    else:
                         _notify_res = await _notify_adapter.send(
                             source.chat_id,
                             _heartbeat_text,