common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(signal): back off sendTyping spam for unreachable recipients (#12118)

Browse Source 
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.
This commit is contained in:
Teknium
2026-04-18 04:13:32 -07:00
committed by GitHub
parent cf012a05d8
commit 9527707f80
2 changed files with 215 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
gateway/platforms/signal.py
View File

@@ -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

137
tests/gateway/test_signal.py
View File

@@ -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