Fix paperclip recall fan-out and Signal typing retry spam

gateway/platforms/signal.py

@@ -548,7 +548,15 @@ async def _fetch_attachment(self, attachment_id: str) -> tuple:
     # JSON-RPC Communication
     # ------------------------------------------------------------------
 
-    async def _rpc(self, method: str, params: dict, rpc_id: str = None) -> Any:
+    async def _rpc(
+        self,
+        method: str,
+        params: dict,
+        rpc_id: str = None,
+        *,
+        log_failures: bool = True,
+        success_on_null_result: bool = False,
+    ) -> Any:
         """Send a JSON-RPC 2.0 request to signal-cli daemon."""
         if not self.client:
             logger.warning("Signal: RPC called but client not connected")
@@ -574,13 +582,22 @@ async def _rpc(self, method: str, params: dict, rpc_id: str = None) -> Any:
             data = resp.json()
 
             if "error" in data:
-                logger.warning("Signal RPC error (%s): %s", method, data["error"])
+                if log_failures:
+                    logger.warning("Signal RPC error (%s): %s", method, data["error"])
+                else:
+                    logger.debug("Signal RPC error (%s): %s", method, data["error"])
                 return None
 
-            return data.get("result")
+            result = data.get("result")
+            if result is None and success_on_null_result:
+                return True
+            return result
 
         except Exception as e:
-            logger.warning("Signal RPC %s failed: %s", method, e)
+            if log_failures:
+                logger.warning("Signal RPC %s failed: %s", method, e)
+            else:
+                logger.debug("Signal RPC %s failed: %s", method, e)
             return None
 
     # ------------------------------------------------------------------
@@ -627,7 +644,11 @@ def _track_sent_timestamp(self, rpc_result) -> None:
                 self._recent_sent_timestamps.pop()
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
-        """Send a typing indicator."""
+        """Start a throttled typing indicator loop for a chat."""
+        existing = self._typing_tasks.get(chat_id)
+        if existing and not existing.done():
+            return
+
         params: Dict[str, Any] = {
             "account": self.account,
         }
@@ -637,7 +658,34 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
         else:
             params["recipient"] = [chat_id]
 
-        await self._rpc("sendTyping", params, rpc_id="typing")
+        async def _typing_loop() -> None:
+            try:
+                while True:
+                    result = await self._rpc(
+                        "sendTyping",
+                        params,
+                        rpc_id="typing",
+                        log_failures=False,
+                        success_on_null_result=True,
+                    )
+                    if result is None:
+                        # Hold the slot through one refresh window so the base
+                        # adapter doesn't recreate a failed typing request every
+                        # two seconds while Signal transport is unhealthy.
+                        await asyncio.sleep(TYPING_INTERVAL)
+                        return
+                    await asyncio.sleep(TYPING_INTERVAL)
+            except asyncio.CancelledError:
+                pass
+
+        task = asyncio.create_task(_typing_loop())
+        self._typing_tasks[chat_id] = task
+
+        def _cleanup(done_task: asyncio.Task) -> None:
+            if self._typing_tasks.get(chat_id) is done_task:
+                self._typing_tasks.pop(chat_id, None)
+
+        task.add_done_callback(_cleanup)
 
     async def send_image(
         self,

tests/gateway/test_signal.py

@@ -1,4 +1,5 @@
 """Tests for Signal messenger platform adapter."""
+import asyncio
 import base64
 import json
 import pytest
@@ -740,3 +741,102 @@ async def test_stop_typing_calls_private_method(self, monkeypatch):
         await adapter.stop_typing("+155****4567")
 
         adapter._stop_typing_indicator.assert_awaited_once_with("+155****4567")
+
+
+class TestSignalTypingLoop:
+    @pytest.mark.asyncio
+    async def test_send_typing_reuses_existing_background_loop(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch)
+        calls = []
+
+        async def _fake_rpc(
+            method,
+            params,
+            rpc_id=None,
+            log_failures=True,
+            success_on_null_result=False,
+        ):
+            calls.append({
+                "method": method,
+                "rpc_id": rpc_id,
+                "log_failures": log_failures,
+                "success_on_null_result": success_on_null_result,
+            })
+            return {"ok": True}
+
+        adapter._rpc = _fake_rpc
+
+        await adapter.send_typing("+155****4567")
+        await adapter.send_typing("+155****4567")
+        await asyncio.sleep(0)
+
+        assert len(adapter._typing_tasks) == 1
+        assert len(calls) == 1
+        assert calls[0]["method"] == "sendTyping"
+        assert calls[0]["log_failures"] is False
+        assert calls[0]["success_on_null_result"] is True
+
+        await adapter.stop_typing("+155****4567")
+        assert "+155****4567" not in adapter._typing_tasks
+
+    @pytest.mark.asyncio
+    async def test_send_typing_treats_null_result_as_success(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch)
+        calls = []
+
+        async def _fake_rpc(
+            method,
+            params,
+            rpc_id=None,
+            log_failures=True,
+            success_on_null_result=False,
+        ):
+            calls.append(success_on_null_result)
+            return True if success_on_null_result else None
+
+        adapter._rpc = _fake_rpc
+
+        await adapter.send_typing("+155****4567")
+        await asyncio.sleep(0.01)
+
+        assert len(calls) == 1
+        assert "+155****4567" in adapter._typing_tasks
+
+        await adapter.stop_typing("+155****4567")
+
+    @pytest.mark.asyncio
+    async def test_send_typing_failure_holds_slot_for_refresh_window(self, monkeypatch):
+        adapter = _make_signal_adapter(monkeypatch)
+        monkeypatch.setattr("gateway.platforms.signal.TYPING_INTERVAL", 0.05)
+        calls = []
+
+        async def _fake_rpc(
+            method,
+            params,
+            rpc_id=None,
+            log_failures=True,
+            success_on_null_result=False,
+        ):
+            calls.append({
+                "method": method,
+                "rpc_id": rpc_id,
+                "log_failures": log_failures,
+                "success_on_null_result": success_on_null_result,
+            })
+            return None
+
+        adapter._rpc = _fake_rpc
+
+        await adapter.send_typing("+155****4567")
+        await asyncio.sleep(0.01)
+        await adapter.send_typing("+155****4567")
+
+        assert len(calls) == 1
+
+        await asyncio.sleep(0.07)
+        await adapter.send_typing("+155****4567")
+        await asyncio.sleep(0.01)
+
+        assert len(calls) == 2
+
+        await adapter.stop_typing("+155****4567")

tests/tools/test_session_search.py

@@ -1,12 +1,14 @@
 """Tests for tools/session_search_tool.py — helper functions and search dispatcher."""
 
+import asyncio
 import json
 import time
 import pytest
 
 from tools.session_search_tool import (
     _format_timestamp,
     _format_conversation,
+    _summarize_session,
     _truncate_around_matches,
     _HIDDEN_SESSION_SOURCES,
     MAX_SESSION_CHARS,
@@ -375,3 +377,90 @@ def _get_session(session_id):
         assert result["count"] == 0
         assert result["results"] == []
         assert result["sessions_searched"] == 0
+
+    def test_serializes_session_summaries_to_avoid_provider_fanout(self):
+        """Session summaries should run one-at-a-time to avoid rate-limit bursts."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = [
+            {"session_id": "s1", "content": "match", "source": "cli", "session_started": 1709500000, "model": "test"},
+            {"session_id": "s2", "content": "match", "source": "cli", "session_started": 1709400000, "model": "test"},
+        ]
+        mock_db.get_session.side_effect = lambda session_id: {"parent_session_id": None}
+        mock_db.get_messages_as_conversation.side_effect = lambda session_id: [
+            {"role": "user", "content": f"hello from {session_id}"},
+        ]
+
+        state = {"current": 0, "max": 0}
+
+        async def _fake_summarize(text, query, meta):
+            state["current"] += 1
+            state["max"] = max(state["max"], state["current"])
+            await asyncio.sleep(0)
+            state["current"] -= 1
+            return f"summary for {query}"
+
+        from unittest.mock import patch as _patch
+        with _patch("tools.session_search_tool._summarize_session", new=_fake_summarize):
+            result = json.loads(session_search(query="test", db=mock_db, limit=2))
+
+        assert result["success"] is True
+        assert result["count"] == 2
+        assert state["max"] == 1
+
+    def test_returns_partial_results_when_serial_summary_budget_expires(self, monkeypatch):
+        """Expired summary budget should fall back to previews instead of failing the whole tool."""
+        from unittest.mock import MagicMock
+        from tools.session_search_tool import session_search
+
+        mock_db = MagicMock()
+        mock_db.search_messages.return_value = [
+            {"session_id": "s1", "content": "match", "source": "cli", "session_started": 1709500000, "model": "test"},
+            {"session_id": "s2", "content": "match", "source": "cli", "session_started": 1709400000, "model": "test"},
+        ]
+        mock_db.get_session.side_effect = lambda session_id: {"parent_session_id": None}
+        mock_db.get_messages_as_conversation.side_effect = lambda session_id: [
+            {"role": "user", "content": f"hello from {session_id}"},
+        ]
+        monkeypatch.setattr(
+            "tools.session_search_tool.SESSION_SEARCH_TOTAL_TIMEOUT_SECONDS",
+            0.01,
+        )
+
+        async def _slow_summarize(text, query, meta):
+            await asyncio.sleep(0.02)
+            return f"summary for {query}"
+
+        from unittest.mock import patch as _patch
+        with _patch("tools.session_search_tool._summarize_session", new=_slow_summarize):
+            result = json.loads(session_search(query="test", db=mock_db, limit=2))
+
+        assert result["success"] is True
+        assert result["count"] == 2
+        assert all("[Raw preview" in entry["summary"] for entry in result["results"])
+
+
+class TestSummarizeSession:
+    @pytest.mark.asyncio
+    async def test_rate_limit_skips_retries(self):
+        class _RateLimitError(Exception):
+            status_code = 429
+
+        calls = {"count": 0}
+
+        async def _rate_limited(*args, **kwargs):
+            calls["count"] += 1
+            raise _RateLimitError("Too Many Requests")
+
+        from unittest.mock import patch as _patch
+        with _patch("tools.session_search_tool.async_call_llm", new=_rate_limited):
+            result = await _summarize_session(
+                "user asked about paperclip adapter loops",
+                "paperclip adapter",
+                {"source": "cli", "started_at": 1709500000},
+            )
+
+        assert result is None
+        assert calls["count"] == 1

tools/session_search_tool.py

@@ -224,6 +224,14 @@ async def _summarize_session(
             logging.warning("No auxiliary model available for session summarization")
             return None
         except Exception as e:
+            status_code = getattr(e, "status_code", None)
+            error_text = str(e).lower()
+            if status_code == 429 or "too many requests" in error_text:
+                logging.warning(
+                    "Session summarization rate-limited; skipping retries for this session: %s",
+                    e,
+                )
+                return None
             if attempt < max_retries - 1:
                 await asyncio.sleep(1 * (attempt + 1))
             else:
@@ -240,6 +248,7 @@ async def _summarize_session(
 # Third-party integrations (Paperclip agents, etc.) tag their sessions with
 # HERMES_SESSION_SOURCE=tool so they don't clutter the user's session history.
 _HIDDEN_SESSION_SOURCES = ("tool",)
+SESSION_SEARCH_TOTAL_TIMEOUT_SECONDS = 60.0
 
 
 def _list_recent_sessions(db, limit: int, current_session_id: str = None) -> str:
@@ -421,14 +430,34 @@ def _resolve_to_parent(session_id: str) -> str:
                     exc_info=True,
                 )
 
-        # Summarize all sessions in parallel
+        # Summarize sessions serially. Parallel fan-out causes auxiliary model
+        # throttling for integrations that trigger recall repeatedly, which can
+        # turn one tool call into a long retry loop.
         async def _summarize_all() -> List[Union[str, Exception]]:
-            """Summarize all sessions in parallel."""
-            coros = [
-                _summarize_session(text, query, meta)
-                for _, _, text, meta in tasks
-            ]
-            return await asyncio.gather(*coros, return_exceptions=True)
+            """Summarize prepared sessions one at a time within a bounded budget."""
+            results: List[Union[str, Exception]] = []
+            loop = asyncio.get_running_loop()
+            deadline = loop.time() + SESSION_SEARCH_TOTAL_TIMEOUT_SECONDS
+
+            for index, (_, _, text, meta) in enumerate(tasks):
+                remaining = deadline - loop.time()
+                if remaining <= 0:
+                    timeout_exc = TimeoutError(
+                        f"Session summarization exceeded {SESSION_SEARCH_TOTAL_TIMEOUT_SECONDS:.0f}s budget"
+                    )
+                    results.append(timeout_exc)
+                    results.extend([timeout_exc] * (len(tasks) - index - 1))
+                    break
+                try:
+                    results.append(
+                        await asyncio.wait_for(
+                            _summarize_session(text, query, meta),
+                            timeout=remaining,
+                        )
+                    )
+                except Exception as exc:
+                    results.append(exc)
+            return results
 
         try:
             # Use _run_async() which properly manages event loops across
@@ -441,12 +470,15 @@ async def _summarize_all() -> List[Union[str, Exception]]:
             results = _run_async(_summarize_all())
         except concurrent.futures.TimeoutError:
             logging.warning(
-                "Session summarization timed out after 60 seconds",
+                "Session summarization timed out after %.0f seconds",
+                SESSION_SEARCH_TOTAL_TIMEOUT_SECONDS,
                 exc_info=True,
             )
             return json.dumps({
                 "success": False,
-                "error": "Session summarization timed out. Try a more specific query or reduce the limit.",
+                "error": (
+                    "Session summarization timed out. Try a more specific query or reduce the limit."
+                ),
             }, ensure_ascii=False)
 
         summaries = []