From 9527707f805a35377169616fa41dd7711e42a9dc Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 18 Apr 2026 04:13:32 -0700 Subject: [PATCH] fix(signal): back off sendTyping spam for unreachable recipients (#12118) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit base.py's _keep_typing refresh loop calls send_typing every ~2s while the agent is processing. If signal-cli returns NETWORK_FAILURE for the recipient (offline, unroutable, group membership lost), the unmitigated path was a WARNING log every 2 seconds for as long as the agent stayed busy — a user report showed 1048 warnings in 41 minutes for one offline contact, plus the matching volume of pointless RPC traffic to signal-cli. - _rpc() accepts log_failures=False so callers can route repeated expected failures (typing) to DEBUG while keeping send/receive at WARNING. - send_typing() tracks consecutive failures per chat. First failure still logs WARNING so transport issues remain visible; subsequent failures log at DEBUG. After three consecutive failures we skip the RPC during an exponential cooldown (16s, 32s, 60s cap) so we stop hammering signal-cli for a recipient it can't deliver to. A successful sendTyping resets the counters. - _stop_typing_indicator() clears the backoff state so the next agent turn starts fresh. E2E simulation against the reported 41-minute window: RPCs drop from 1230 to 45 (-96%), log lines from 1048 WARNINGs to 1 WARNING + 44 DEBUGs. Credits kshitijk4poor (#12056) for the _rpc log_failures kwarg idea; the broader restructure in that PR (nested per-chat loop inside send_typing) is avoided here in favour of stateful backoff that preserves base.py's existing _keep_typing architecture. --- gateway/platforms/signal.py | 84 +++++++++++++++++++-- tests/gateway/test_signal.py | 137 +++++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+), 6 deletions(-) diff --git a/gateway/platforms/signal.py b/gateway/platforms/signal.py index 617713ad9..4df4193bc 100644 --- a/gateway/platforms/signal.py +++ b/gateway/platforms/signal.py @@ -160,6 +160,14 @@ class SignalAdapter(BasePlatformAdapter): self._sse_task: Optional[asyncio.Task] = None self._health_monitor_task: Optional[asyncio.Task] = None self._typing_tasks: Dict[str, asyncio.Task] = {} + # Per-chat typing-indicator backoff. When signal-cli reports + # NETWORK_FAILURE (recipient offline / unroutable), base.py's + # _keep_typing refresh loop would otherwise hammer sendTyping every + # ~2s indefinitely, producing WARNING-level log spam and pointless + # RPC traffic. We track consecutive failures per chat and skip the + # RPC during a cooldown window instead. + self._typing_failures: Dict[str, int] = {} + self._typing_skip_until: Dict[str, float] = {} self._running = False self._last_sse_activity = 0.0 self._sse_response: Optional[httpx.Response] = None @@ -548,8 +556,22 @@ class SignalAdapter(BasePlatformAdapter): # JSON-RPC Communication # ------------------------------------------------------------------ - async def _rpc(self, method: str, params: dict, rpc_id: str = None) -> Any: - """Send a JSON-RPC 2.0 request to signal-cli daemon.""" + async def _rpc( + self, + method: str, + params: dict, + rpc_id: str = None, + *, + log_failures: bool = True, + ) -> Any: + """Send a JSON-RPC 2.0 request to signal-cli daemon. + + When ``log_failures=False``, error and exception paths log at DEBUG + instead of WARNING — used by the typing-indicator path to silence + repeated NETWORK_FAILURE spam for unreachable recipients while + still preserving visibility for the first occurrence and for + unrelated RPCs. + """ if not self.client: logger.warning("Signal: RPC called but client not connected") return None @@ -574,13 +596,19 @@ class SignalAdapter(BasePlatformAdapter): 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") 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 +655,28 @@ class SignalAdapter(BasePlatformAdapter): self._recent_sent_timestamps.pop() async def send_typing(self, chat_id: str, metadata=None) -> None: - """Send a typing indicator.""" + """Send a typing indicator. + + base.py's ``_keep_typing`` refresh loop calls this every ~2s while + the agent is processing. If signal-cli returns NETWORK_FAILURE for + this recipient (offline, unroutable, group membership lost, etc.) + the unmitigated behaviour is: a WARNING log every 2 seconds for as + long as the agent keeps running. Instead we: + + - silence the WARNING after the first consecutive failure (subsequent + attempts log at DEBUG) so transport issues are still visible once + but don't flood the log, + - skip the RPC entirely during an exponential cooldown window once + three consecutive failures have happened, so we stop hammering + signal-cli with requests it can't deliver. + + A successful sendTyping clears the counters. + """ + now = time.monotonic() + skip_until = self._typing_skip_until.get(chat_id, 0.0) + if now < skip_until: + return + params: Dict[str, Any] = { "account": self.account, } @@ -637,7 +686,26 @@ class SignalAdapter(BasePlatformAdapter): else: params["recipient"] = [chat_id] - await self._rpc("sendTyping", params, rpc_id="typing") + fails = self._typing_failures.get(chat_id, 0) + result = await self._rpc( + "sendTyping", + params, + rpc_id="typing", + log_failures=(fails == 0), + ) + + if result is None: + fails += 1 + self._typing_failures[chat_id] = fails + # After 3 consecutive failures, back off exponentially (16s, + # 32s, 60s cap) to stop spamming signal-cli for a recipient + # that clearly isn't reachable right now. + if fails >= 3: + backoff = min(60.0, 16.0 * (2 ** (fails - 3))) + self._typing_skip_until[chat_id] = now + backoff + else: + self._typing_failures.pop(chat_id, None) + self._typing_skip_until.pop(chat_id, None) async def send_image( self, @@ -789,6 +857,10 @@ class SignalAdapter(BasePlatformAdapter): await task except asyncio.CancelledError: pass + # Reset per-chat typing backoff state so the next agent turn starts + # fresh rather than inheriting a cooldown from a prior conversation. + self._typing_failures.pop(chat_id, None) + self._typing_skip_until.pop(chat_id, None) async def stop_typing(self, chat_id: str) -> None: """Public interface for stopping typing — called by base adapter's diff --git a/tests/gateway/test_signal.py b/tests/gateway/test_signal.py index 26f1e4f3b..eee3a0db8 100644 --- a/tests/gateway/test_signal.py +++ b/tests/gateway/test_signal.py @@ -740,3 +740,140 @@ class TestSignalStopTyping: await adapter.stop_typing("+155****4567") adapter._stop_typing_indicator.assert_awaited_once_with("+155****4567") + + +# --------------------------------------------------------------------------- +# Typing-indicator backoff on repeated failures (Signal RPC spam fix) +# --------------------------------------------------------------------------- + +class TestSignalTypingBackoff: + """When base.py's _keep_typing refresh loop calls send_typing every ~2s + and the recipient is unreachable (NETWORK_FAILURE), the adapter must: + + - log WARNING only for the first failure (subsequent failures use DEBUG + via log_failures=False on the _rpc call) + - after 3 consecutive failures, skip the RPC entirely during an + exponential cooldown window instead of hammering signal-cli every 2s + - reset counters on a successful sendTyping + - reset counters when _stop_typing_indicator() is called for the chat + """ + + @pytest.mark.asyncio + async def test_first_failure_logs_at_warning_subsequent_at_debug( + self, monkeypatch + ): + adapter = _make_signal_adapter(monkeypatch) + calls = [] + + async def _fake_rpc(method, params, rpc_id=None, *, log_failures=True): + calls.append({"log_failures": log_failures}) + return None # simulate NETWORK_FAILURE + + adapter._rpc = _fake_rpc + + await adapter.send_typing("+155****4567") + await adapter.send_typing("+155****4567") + + assert len(calls) == 2 + assert calls[0]["log_failures"] is True # first failure — warn + assert calls[1]["log_failures"] is False # subsequent — debug + + @pytest.mark.asyncio + async def test_three_consecutive_failures_trigger_cooldown( + self, monkeypatch + ): + adapter = _make_signal_adapter(monkeypatch) + call_count = {"n": 0} + + async def _fake_rpc(method, params, rpc_id=None, *, log_failures=True): + call_count["n"] += 1 + return None + + adapter._rpc = _fake_rpc + + # Three failures engage the cooldown. + await adapter.send_typing("+155****4567") + await adapter.send_typing("+155****4567") + await adapter.send_typing("+155****4567") + assert call_count["n"] == 3 + assert "+155****4567" in adapter._typing_skip_until + + # Fourth, fifth, ... calls during the cooldown window are short- + # circuited — the RPC is not issued at all. + await adapter.send_typing("+155****4567") + await adapter.send_typing("+155****4567") + assert call_count["n"] == 3 + + @pytest.mark.asyncio + async def test_cooldown_is_per_chat_not_global(self, monkeypatch): + adapter = _make_signal_adapter(monkeypatch) + call_log = [] + + async def _fake_rpc(method, params, rpc_id=None, *, log_failures=True): + call_log.append(params.get("recipient") or params.get("groupId")) + return None + + adapter._rpc = _fake_rpc + + # Drive chat A into cooldown. + for _ in range(3): + await adapter.send_typing("+155****4567") + assert "+155****4567" in adapter._typing_skip_until + + # Chat B is unaffected — still makes RPCs. + await adapter.send_typing("+155****9999") + await adapter.send_typing("+155****9999") + assert "+155****9999" not in adapter._typing_skip_until + # Chat A cooldown untouched + assert "+155****4567" in adapter._typing_skip_until + + @pytest.mark.asyncio + async def test_success_resets_failure_counter_and_cooldown( + self, monkeypatch + ): + adapter = _make_signal_adapter(monkeypatch) + result_queue = [None, None, {"timestamp": 12345}] + call_log = [] + + async def _fake_rpc(method, params, rpc_id=None, *, log_failures=True): + call_log.append(log_failures) + return result_queue.pop(0) + + adapter._rpc = _fake_rpc + + await adapter.send_typing("+155****4567") # fail 1 — warn + await adapter.send_typing("+155****4567") # fail 2 — debug + await adapter.send_typing("+155****4567") # success — reset + + assert adapter._typing_failures.get("+155****4567", 0) == 0 + assert "+155****4567" not in adapter._typing_skip_until + + # Next failure after recovery logs at WARNING again (fresh counter). + async def _fail(method, params, rpc_id=None, *, log_failures=True): + call_log.append(log_failures) + return None + + adapter._rpc = _fail + await adapter.send_typing("+155****4567") + assert call_log[-1] is True # first failure in a fresh cycle + + @pytest.mark.asyncio + async def test_stop_typing_indicator_clears_backoff_state( + self, monkeypatch + ): + adapter = _make_signal_adapter(monkeypatch) + + async def _fail(method, params, rpc_id=None, *, log_failures=True): + return None + + adapter._rpc = _fail + + for _ in range(3): + await adapter.send_typing("+155****4567") + assert adapter._typing_failures.get("+155****4567") == 3 + assert "+155****4567" in adapter._typing_skip_until + + await adapter._stop_typing_indicator("+155****4567") + + assert "+155****4567" not in adapter._typing_failures + assert "+155****4567" not in adapter._typing_skip_until