# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """Whether the bot can still hear, and the gate that stops it saying yes anyway. On 2026-08-29 signal-cli's connection to Signal died and reconnected in a loop for three hours. Every surface said the bot was fine: the container's own healthcheck answered 204, ``/v1/groups`` answered 200 out of local state, ``/healthz`` answered 200, and the bot's WebSocket to the container stayed *open* -- the container keeps client sockets alive even when its own upstream is dead, so the transport's reconnect logic never fired. The only symptom was silence, and silence is also what a quiet evening looks like. So this module answers "is the silence real?" with a question nobody can answer from a cache: ``/v1/devices``, which needs a live round trip to Signal. It is asked only after the silence has gone on long enough to be worth asking about, so a busy group costs nothing and an idle one costs one request per silence window. """ from __future__ import annotations import asyncio import logging import time from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING, Final import httpx if TYPE_CHECKING: from signal_bot.config import Config logger = logging.getLogger(__name__) # Measured against the real failure: the broken endpoint answered in 3.9s # once and hung past 20s another time. 15s is above the observed failure # latency and below config.liveness_tick_seconds, so probes cannot pile up. PROBE_TIMEOUT_SECONDS: Final = 15.0 # signal-cli's own reconnect loop backs off in 100..3200ms, so the failure is # stable within a second. Three probes 30s apart is not "waiting it out", it # is refusing to trip on one bad round trip. FAILURES_BEFORE_DEAF: Final = 3 Ticker = Callable[[], float] Sleeper = Callable[[float], Awaitable[None]] class Liveness: """Tracks proof that messages can still arrive, and reports when they cannot.""" def __init__( self, config: Config, client: httpx.AsyncClient, *, ticker: Ticker | None = None, sleeper: Sleeper | None = None, failures_before_deaf: int = FAILURES_BEFORE_DEAF, ) -> None: """Wire the gate to its config and its injectable I/O. Args: config: Resolved runtime configuration. client: HTTP client used for the probe. ticker: Monotonic clock. Injected by the tests. sleeper: Async sleep used by run(). Injected by the tests. failures_before_deaf: Consecutive failed probes before the bot stops calling itself healthy. """ self._config = config self._client = client self._now = ticker if ticker is not None else time.monotonic self._sleep = sleeper if sleeper is not None else asyncio.sleep self._limit = failures_before_deaf # Healthy until proven otherwise: the installer waits on /healthz # with a 60s budget, and a cold "unknown means 503" would fail the # install and burn the unit's restart limit for no reason. self._last_alive = self._now() self._failures = 0 self._detail = "" @property def healthy(self) -> bool: """True while the bot has no evidence that it has gone deaf.""" return self._failures < self._limit @property def silent_for(self) -> float: """Seconds since the last envelope or successful probe.""" return self._now() - self._last_alive @property def reason(self) -> str: """Why the bot believes it is deaf, in one line for /healthz.""" return ( f"nothing received for {self.silent_for:.0f}s and {self._failures} " f"consecutive failed probes of {self._config.devices_url}: {self._detail}" ) def note_envelope(self) -> None: """Record proof of life: any frame at all, or a successful probe.""" self._last_alive = self._now() if self._failures: logger.warning("receive path recovered after %d failed probe(s)", self._failures) self._failures = 0 self._detail = "" async def probe(self) -> bool: """Ask signal-cli something it cannot answer from local state. Returns: True when the account's connection to Signal is alive. """ try: response = await self._client.get( self._config.devices_url, timeout=PROBE_TIMEOUT_SECONDS ) except httpx.HTTPError as exc: self._detail = f"{type(exc).__name__}: {exc}" return False if response.status_code != httpx.codes.OK: self._detail = f"HTTP {response.status_code}: {response.text.strip()[:200]}" return False return True async def run(self, iterations: int | None = None) -> None: """Check for deafness forever. Args: iterations: Stop after this many ticks. None means forever. """ completed = 0 while iterations is None or completed < iterations: await self._sleep(self._config.liveness_tick_seconds) await self.check() completed += 1 async def check(self) -> None: """Probe once, but only when the silence has lasted long enough. Public because it is the whole gate in one call: the tests, and any future hand-run diagnostic, must be able to ask the same question the loop asks without reaching into private state. """ silent = self.silent_for if silent < self._config.silence_seconds: return if await self.probe(): # A live probe is proof of life exactly like an envelope, and # resetting the clock here is what bounds the cost: an idle # account probes once per silence window, not once per tick. self.note_envelope() return self._failures += 1 if self._failures <= self._limit: logger.warning( "signal-cli cannot reach Signal (failure %d of %d) after %.0fs of silence: %s", self._failures, self._limit, silent, self._detail, ) return # Past the limit the verdict is already in, and "failure 8 of 3" # reads like a bug in the counter rather than a bot that is still # deaf. Keep saying it, but say it truthfully. logger.warning( "still deaf: %d failed probe(s) and %.0fs of silence: %s", self._failures, silent, self._detail, )