feat(telegram): fire telegram:reaction hook when user reacts to bot messages

agent/agent_runtime_helpers.py

@@ -2047,6 +2047,46 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
         filtered.append(msg)
     messages = filtered
 
+    # --- Strip tool_calls from compaction-summary assistant messages ---
+    # When context compression fires, the compressed head may contain an
+    # assistant message that (a) carries the compaction summary as its text
+    # content AND (b) still has tool_calls from the turn that was compressed.
+    # Those tool_calls have no matching tool results in the active message list
+    # (the results were archived together with the pre-compaction transcript).
+    # Anthropic rejects the resulting wire messages with HTTP 400
+    # "tool_use ids found without tool_result blocks immediately after".
+    # Fix: strip tool_calls from any assistant message whose content is a
+    # compaction summary — they are historical artifacts that the model must
+    # not act on.
+    _COMPACTION_MARKERS = (
+        "[CONTEXT COMPACTION — REFERENCE ONLY]",
+        "[CONTEXT SUMMARY]:",
+    )
+    stripped_compaction_calls = 0
+    for i, msg in enumerate(messages):
+        if msg.get("role") != "assistant":
+            continue
+        if not msg.get("tool_calls"):
+            continue
+        content = msg.get("content", "")
+        if isinstance(content, list):
+            # Multimodal content: check text blocks
+            content = " ".join(
+                b.get("text", "") for b in content
+                if isinstance(b, dict) and b.get("type") == "text"
+            )
+        if isinstance(content, str) and any(
+            content.lstrip().startswith(m) for m in _COMPACTION_MARKERS
+        ):
+            messages[i] = {**msg}
+            messages[i].pop("tool_calls", None)
+            stripped_compaction_calls += 1
+    if stripped_compaction_calls:
+        _ra().logger.debug(
+            "Pre-call sanitizer: stripped tool_calls from %d compaction-summary message(s)",
+            stripped_compaction_calls,
+        )
+
     surviving_call_ids: set = set()
     for msg in messages:
         if msg.get("role") == "assistant":

agent/anthropic_adapter.py

@@ -2013,57 +2013,84 @@ def _strip_orphaned_tool_blocks(result: List[Dict[str, Any]]) -> None:
     """Strip tool_use blocks with no matching tool_result, and vice versa.
 
     Context compression or session truncation can remove either side of a
-    tool-call pair.  Anthropic rejects both orphans with HTTP 400.
-
+    tool-call pair, or insert messages between a tool_use and its result.
+    Anthropic requires each tool_use to have a matching tool_result in the
+    IMMEDIATELY FOLLOWING user message — a global ID match is not enough.
     Mutates ``result`` in place.
     """
-    # Strip orphaned tool_use blocks (no matching tool_result follows)
-    tool_result_ids = set()
-    for m in result:
-        if m["role"] == "user" and isinstance(m["content"], list):
-            for block in m["content"]:
-                if block.get("type") == "tool_result":
-                    tool_result_ids.add(block.get("tool_use_id"))
-    for m in result:
-        if m["role"] == "assistant" and isinstance(m["content"], list):
-            kept = [
-                b
-                for b in m["content"]
-                if b.get("type") != "tool_use" or b.get("id") in tool_result_ids
-            ]
-            # If stripping an orphaned tool_use mutated a turn that also carries a
-            # signed thinking block, that block's Anthropic signature was computed
-            # against the ORIGINAL (un-stripped) turn content and is now invalid.
-            # Anthropic rejects the replayed turn with HTTP 400 "thinking blocks in
-            # the latest assistant message cannot be modified".  Flag the turn so
-            # _manage_thinking_signatures can demote the dead signature instead of
-            # replaying it verbatim.  See hermes-agent: extended-thinking + parallel
-            # tool batch interrupted mid-flight → non-retryable 400 crash-loop.
-            if len(kept) != len(m["content"]) and any(
-                isinstance(b, dict) and b.get("type") in {"thinking", "redacted_thinking"}
-                for b in m["content"]
-            ):
-                m["_thinking_signature_invalidated"] = True
-            m["content"] = kept
-            if not m["content"]:
-                m["content"] = [{"type": "text", "text": "(tool call removed)"}]
-
-    # Strip orphaned tool_result blocks (no matching tool_use precedes them)
-    tool_use_ids = set()
+    # Pass 1: for each assistant turn, collect only the tool_result IDs from
+    # the IMMEDIATELY FOLLOWING user message (adjacent_result_ids).
+    # Strip tool_use blocks not covered by that adjacent set.
+    for i, m in enumerate(result):
+        if m.get("role") != "assistant" or not isinstance(m.get("content"), list):
+            continue
+
+        tool_use_ids_in_turn = {
+            b.get("id")
+            for b in m["content"]
+            if isinstance(b, dict) and b.get("type") == "tool_use"
+        }
+        if not tool_use_ids_in_turn:
+            continue
+
+        # Find the immediately following user message (skip non-dict entries)
+        next_user = None
+        for j in range(i + 1, len(result)):
+            if isinstance(result[j], dict) and result[j].get("role") == "user":
+                next_user = result[j]
+                break
+
+        adjacent_result_ids: set = set()
+        if next_user is not None and isinstance(next_user.get("content"), list):
+            for block in next_user["content"]:
+                if isinstance(block, dict) and block.get("type") == "tool_result":
+                    adjacent_result_ids.add(block.get("tool_use_id"))
+
+        orphaned = tool_use_ids_in_turn - adjacent_result_ids
+        if not orphaned:
+            continue
+
+        kept = [
+            b
+            for b in m["content"]
+            if not (isinstance(b, dict) and b.get("type") == "tool_use" and b.get("id") in orphaned)
+        ]
+        # If stripping an orphaned tool_use mutated a turn that also carries a
+        # signed thinking block, that block's Anthropic signature was computed
+        # against the ORIGINAL (un-stripped) turn content and is now invalid.
+        # Anthropic rejects the replayed turn with HTTP 400 "thinking blocks in
+        # the latest assistant message cannot be modified".  Flag the turn so
+        # _manage_thinking_signatures can demote the dead signature instead of
+        # replaying it verbatim.  See hermes-agent: extended-thinking + parallel
+        # tool batch interrupted mid-flight → non-retryable 400 crash-loop.
+        if len(kept) != len(m["content"]) and any(
+            isinstance(b, dict) and b.get("type") in {"thinking", "redacted_thinking"}
+            for b in m["content"]
+        ):
+            m["_thinking_signature_invalidated"] = True
+        m["content"] = kept if kept else [{"type": "text", "text": "(tool call removed)"}]
+
+    # Pass 2: Rebuild the set of tool_use IDs that survived pass 1, then
+    # strip tool_result blocks that no longer have any matching tool_use
+    # anywhere in the conversation.
+    surviving_tool_use_ids: set = set()
     for m in result:
-        if m["role"] == "assistant" and isinstance(m["content"], list):
+        if m.get("role") == "assistant" and isinstance(m.get("content"), list):
             for block in m["content"]:
-                if block.get("type") == "tool_use":
-                    tool_use_ids.add(block.get("id"))
+                if isinstance(block, dict) and block.get("type") == "tool_use":
+                    surviving_tool_use_ids.add(block.get("id"))
+
     for m in result:
-        if m["role"] == "user" and isinstance(m["content"], list):
-            m["content"] = [
-                b
-                for b in m["content"]
-                if b.get("type") != "tool_result" or b.get("tool_use_id") in tool_use_ids
-            ]
-            if not m["content"]:
-                m["content"] = [{"type": "text", "text": "(tool result removed)"}]
+        if m.get("role") != "user" or not isinstance(m.get("content"), list):
+            continue
+        new_content = [
+            b
+            for b in m["content"]
+            if not (isinstance(b, dict) and b.get("type") == "tool_result")
+            or b.get("tool_use_id") in surviving_tool_use_ids
+        ]
+        if len(new_content) != len(m["content"]):
+            m["content"] = new_content if new_content else [{"type": "text", "text": "(tool result removed)"}]
 
 
 def _merge_consecutive_roles(result: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
@@ -2315,8 +2342,8 @@ def convert_messages_to_anthropic(
         # Regular user message
         result.append(_convert_user_message(content))
 
-    _strip_orphaned_tool_blocks(result)
     result = _merge_consecutive_roles(result)
+    _strip_orphaned_tool_blocks(result)
     _manage_thinking_signatures(result, base_url, model)
     _evict_old_screenshots(result)
 

agent/conversation_loop.py

@@ -2614,6 +2614,91 @@ def _perform_api_call(next_api_kwargs):
                     )
                     continue
 
+                # ── Orphaned tool_use recovery ─────────────────────────
+                # Anthropic rejects messages where a tool_use block is not
+                # immediately followed by a matching tool_result.  Two
+                # known causes:
+                #   1. Context compression inserts messages between a
+                #      tool_use and its tool_result.
+                #   2. A cron/subagent session is interrupted before the
+                #      tool_result is appended (e.g. execute_code blocked
+                #      by the approval guard).
+                # The canonical ``messages`` list uses OpenAI-style
+                # role=tool / tool_calls — not the Anthropic wire format.
+                # _strip_orphaned_tool_blocks operates on Anthropic-style
+                # api_messages, so we strip there and then signal the
+                # outer loop to rebuild api_messages from the cleaned
+                # canonical list by removing the orphaned tool_calls
+                # entries and their matching role=tool messages.
+                # One-shot to avoid an infinite strip loop.
+                if (
+                    classified.reason == FailoverReason.orphaned_tool_use
+                    and not _retry.orphaned_tool_use_retry_attempted
+                ):
+                    _retry.orphaned_tool_use_retry_attempted = True
+                    try:
+                        # Parse the orphaned tool_use IDs directly from the
+                        # Anthropic error message.  The error looks like:
+                        #   "messages.N: `tool_use` ids were found without
+                        #    `tool_result` blocks immediately after:
+                        #    toolu_xxx, toolu_yyy."
+                        # We cannot rely on detecting orphans from the
+                        # canonical messages (OpenAI-style tool_calls) because
+                        # the pair IS present there — the adjacency breaks
+                        # during Anthropic adapter conversion, e.g. when
+                        # context compaction injects a synthetic user message
+                        # between an assistant tool_use and its tool_result.
+                        import re as _re
+                        _err_str = str(api_error or "")
+                        _orphaned_ids: set = set(
+                            _re.findall(r"toolu_[A-Za-z0-9]+", _err_str)
+                        )
+
+                        # Remove these IDs from canonical messages so the
+                        # next api_messages rebuild produces valid adjacency.
+                        _stripped_canonical = 0
+                        if _orphaned_ids:
+                            _i = 0
+                            while _i < len(messages):
+                                _cm = messages[_i]
+                                if not isinstance(_cm, dict):
+                                    _i += 1
+                                    continue
+                                if _cm.get("role") == "assistant" and isinstance(_cm.get("tool_calls"), list):
+                                    _kept = [
+                                        tc for tc in _cm["tool_calls"]
+                                        if tc.get("id") not in _orphaned_ids
+                                    ]
+                                    if len(_kept) != len(_cm["tool_calls"]):
+                                        _stripped_canonical += len(_cm["tool_calls"]) - len(_kept)
+                                        if _kept:
+                                            _cm["tool_calls"] = _kept
+                                        else:
+                                            _cm.pop("tool_calls", None)
+                                if _cm.get("role") == "tool" and _cm.get("tool_call_id") in _orphaned_ids:
+                                    messages.pop(_i)
+                                    _stripped_canonical += 1
+                                    continue
+                                _i += 1
+                    except Exception as _strip_exc:
+                        logger.warning(
+                            "%sOrphaned tool_use recovery: strip failed: %s",
+                            agent.log_prefix, _strip_exc,
+                        )
+                        _orphaned_ids = set()
+                        _stripped_canonical = 0
+                    agent._vprint(
+                        f"{agent.log_prefix}⚠️  Orphaned tool_use detected — "
+                        f"stripped {len(_orphaned_ids)} id(s) from api_messages "
+                        f"and {_stripped_canonical} canonical entry/entries, retrying...",
+                        force=True,
+                    )
+                    logger.warning(
+                        "%sOrphaned tool_use recovery: stripped ids=%s canonical_entries=%d",
+                        agent.log_prefix, _orphaned_ids, _stripped_canonical,
+                    )
+                    continue
+
                 # ── llama.cpp grammar-parse recovery ──────────────────
                 # llama.cpp's ``json-schema-to-grammar`` converter rejects
                 # regex escape classes (``\d``, ``\w``, ``\s``) and most

agent/error_classifier.py

@@ -52,6 +52,7 @@ class FailoverReason(enum.Enum):
     # Request format
     format_error = "format_error"        # 400 bad request — abort or strip + retry
     invalid_encrypted_content = "invalid_encrypted_content"  # Responses replay blob rejected — strip replay state and retry
+    orphaned_tool_use = "orphaned_tool_use"  # tool_use block has no adjacent tool_result — strip orphans and retry
     multimodal_tool_content_unsupported = "multimodal_tool_content_unsupported"  # Provider rejected list-type content in tool messages (e.g. Xiaomi MiMo) — downgrade to text and retry
 
     # Provider-specific
@@ -1096,6 +1097,18 @@ def _classify_400(
             should_compress=True,
         )
 
+    # Anthropic rejects messages where a tool_use block is not immediately
+    # followed by a matching tool_result.  This can happen when:
+    #   • context compression inserts messages between the pair, or
+    #   • a cron/subagent session is interrupted before the tool_result
+    #     is appended (e.g. execute_code blocked by the approval guard).
+    # Recovery: strip orphaned tool_use/tool_result blocks and retry once.
+    if "tool_use" in error_msg and "tool_result" in error_msg:
+        return result_fn(
+            FailoverReason.orphaned_tool_use,
+            retryable=True,
+        )
+
     # Non-retryable format error
     return result_fn(
         FailoverReason.format_error,

agent/turn_retry_state.py

@@ -49,6 +49,7 @@ class TurnRetryState:
     # ── Format / payload recovery guards ─────────────────────────────────
     thinking_sig_retry_attempted: bool = False
     invalid_encrypted_content_retry_attempted: bool = False
+    orphaned_tool_use_retry_attempted: bool = False
     image_shrink_retry_attempted: bool = False
     multimodal_tool_content_retry_attempted: bool = False
     oauth_1m_beta_retry_attempted: bool = False

gateway/platforms/base.py

@@ -2619,6 +2619,19 @@ def _apply_topic_recovery(self, event: MessageEvent) -> None:
         except Exception:
             logger.debug("topic recovery rewrite failed", exc_info=True)
 
+    def set_reaction_callback(self, callback: Optional[Callable[[dict], Awaitable[None]]]) -> None:
+        """Install a callback that fires when a user reacts to a bot message.
+
+        The callback receives a dict with:
+          - chat_id (str)
+          - message_id (str)
+          - user_id (str | None)
+          - new_reactions (list[str])  — emoji strings added
+          - old_reactions (list[str])  — emoji strings removed
+          - message_text (str | None)  — original bot message text (from rich_sent_store)
+        """
+        self._reaction_callback: Optional[Callable[[dict], Awaitable[None]]] = callback
+
     def set_busy_session_handler(self, handler: Optional[Callable[[MessageEvent, str], Awaitable[bool]]]) -> None:
         """Set an optional handler for messages arriving during active sessions."""
         self._busy_session_handler = handler