feat: PING watchdog, IRCv3 server-time, push notifications

PING watchdog sends PING after configurable silence interval and
disconnects on timeout, detecting stale connections that TCP alone
misses. IRCv3 server-time capability is requested on every connection;
timestamps are injected on dispatch and backlog replay for clients
that support message tags. Push notifications via ntfy or generic
webhook fire on highlights and PMs when no clients are attached,
with configurable cooldown and optional SOCKS5 routing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tests/test_network.py

@@ -4,8 +4,8 @@ from __future__ import annotations
 
 import asyncio
 import base64
-import hashlib
 import random
+import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -13,13 +13,10 @@ import pytest
 from bouncer.config import BouncerConfig, NetworkConfig, ProxyConfig
 from bouncer.irc import IRCMessage, parse
 from bouncer.network import (
-    Network,
-    State,
-    _BIGRAMS,
-    _GENERIC_IDENTS,
-    _GENERIC_REALNAMES,
     _STARTERS,
     _VOWELS,
+    Network,
+    State,
     _markov_word,
     _nick_for_host,
     _password_for_host,
@@ -28,7 +25,6 @@ from bouncer.network import (
     _seeded_markov,
 )
 
-
 # -- helpers -----------------------------------------------------------------
 
 def _cfg(name: str = "testnet", host: str = "irc.test.net", port: int = 6697,
@@ -854,9 +850,11 @@ class TestConnect:
         user_sent = any(b"USER " in c for c in calls)
         assert nick_sent
         assert user_sent
-        # Should NOT have sent CAP REQ sasl
-        cap_sent = any(b"CAP REQ" in c for c in calls)
-        assert not cap_sent
+        # Should have sent CAP REQ server-time but NOT CAP REQ sasl
+        cap_server_time = any(b"CAP REQ server-time" in c for c in calls)
+        cap_sasl = any(b"CAP REQ sasl" in c for c in calls)
+        assert cap_server_time
+        assert not cap_sasl
 
     @pytest.mark.asyncio
     async def test_connect_with_sasl_plain(self) -> None:
@@ -1201,3 +1199,206 @@ class TestReadLoop:
                 await net._read_loop()
 
         assert cb.call_count == 2
+
+
+# -- PING watchdog -----------------------------------------------------------
+
+class TestPingWatchdog:
+    @pytest.mark.asyncio
+    async def test_timeout_triggers_disconnect(self) -> None:
+        """Stale connection triggers disconnect + reconnect."""
+        net = _net(bouncer_cfg=_bouncer(ping_interval=0, ping_timeout=0))
+        net.state = State.READY
+        net._running = True
+        net._last_recv = time.monotonic() - 1000  # stale
+
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+
+        with patch.object(net, "_disconnect", new_callable=AsyncMock) as mock_disc:
+            with patch.object(net, "_schedule_reconnect") as mock_recon:
+                await net._ping_watchdog()
+
+        mock_disc.assert_awaited_once()
+        mock_recon.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_healthy_connection_stays_alive(self) -> None:
+        """Fresh data during timeout window prevents disconnect."""
+        net = _net(bouncer_cfg=_bouncer(ping_interval=10, ping_timeout=5))
+        net.state = State.READY
+        net._running = True
+
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+
+        # Control time progression: stale at first, then fresh after PING
+        clock = [100.0]  # start at T=100
+
+        def fake_monotonic() -> float:
+            return clock[0]
+
+        original_sleep = asyncio.sleep
+        call_count = 0
+
+        async def fake_sleep(delay: float) -> None:
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                # After interval sleep: time advanced, data is stale -> PING sent
+                clock[0] = 120.0
+                net._last_recv = 100.0  # stale (20s > interval=10)
+            elif call_count == 2:
+                # During timeout wait: simulate PONG received (fresh data)
+                clock[0] = 122.0
+                net._last_recv = 122.0  # fresh
+            elif call_count == 3:
+                # Next interval sleep: exit loop
+                net.state = State.DISCONNECTED
+            await original_sleep(0)
+
+        with patch("time.monotonic", side_effect=fake_monotonic):
+            with patch("asyncio.sleep", side_effect=fake_sleep):
+                with patch.object(net, "_disconnect", new_callable=AsyncMock) as mock_disc:
+                    await net._ping_watchdog()
+
+        # PING was sent, but fresh data arrived -- no disconnect
+        ping_sent = any(b"PING" in c.args[0] for c in writer.write.call_args_list)
+        assert ping_sent
+        mock_disc.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_watchdog_cancelled_on_disconnect(self) -> None:
+        net = _net()
+        net.state = State.READY
+        net._running = True
+        net._last_recv = time.monotonic()
+
+        task = MagicMock()
+        task.done.return_value = False
+        net._ping_task = task
+
+        await net._disconnect()
+        task.cancel.assert_called_once()
+        assert net._ping_task is None
+
+    @pytest.mark.asyncio
+    async def test_ping_task_started_in_go_ready(self) -> None:
+        net = _net(bouncer_cfg=_bouncer(probation_seconds=0, ping_interval=999))
+        net.state = State.PROBATION
+        net._running = True
+        net._sasl_complete.set()
+
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+        bl = _mock_backlog()
+        net.backlog = bl
+
+        await net._go_ready()
+        assert net._ping_task is not None
+        # Cancel it so test doesn't leak
+        net._ping_task.cancel()
+        try:
+            await net._ping_task
+        except asyncio.CancelledError:
+            pass
+
+    @pytest.mark.asyncio
+    async def test_ping_task_in_stop(self) -> None:
+        """stop() cancels the ping task."""
+        net = _net()
+        net._running = True
+        ping_task = MagicMock()
+        ping_task.done.return_value = False
+        net._ping_task = ping_task
+
+        await net.stop()
+        # cancel() is called in both stop() and _disconnect()
+        assert ping_task.cancel.call_count >= 1
+
+
+# -- IRCv3 CAP negotiation (server-time) ------------------------------------
+
+class TestCapServerTime:
+    @pytest.mark.asyncio
+    async def test_server_time_ack_sets_flag(self) -> None:
+        net = _net()
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+        net._caps_pending = 1
+
+        await net._handle(_msg(":server CAP * ACK :server-time"))
+        assert net._server_time is True
+
+    @pytest.mark.asyncio
+    async def test_server_time_nak_handled(self) -> None:
+        net = _net()
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+        net._caps_pending = 1
+
+        await net._handle(_msg(":server CAP * NAK :server-time"))
+        assert net._server_time is False
+
+    @pytest.mark.asyncio
+    async def test_combined_sasl_and_server_time(self) -> None:
+        """Both caps requested: ACK server-time + ACK sasl."""
+        net = _net()
+        net._sasl_mechanism = "PLAIN"
+        net._sasl_nick = "nick"
+        net._sasl_pass = "pass"
+        net._caps_pending = 2
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+
+        # ACK server-time first
+        await net._handle(_msg(":server CAP * ACK :server-time"))
+        assert net._server_time is True
+        # Should NOT have sent CAP END yet (SASL still pending)
+        cap_end_calls = [c for c in writer.write.call_args_list
+                         if b"CAP END" in c.args[0]]
+        assert len(cap_end_calls) == 0
+
+        # ACK sasl starts AUTHENTICATE flow
+        await net._handle(_msg(":server CAP * ACK :sasl"))
+        writer.write.assert_called_with(b"AUTHENTICATE PLAIN\r\n")
+
+        # SASL success resolves the last cap
+        await net._handle(_msg(":server 903 nick :SASL authentication successful"))
+        cap_end_calls = [c for c in writer.write.call_args_list
+                         if b"CAP END" in c.args[0]]
+        assert len(cap_end_calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_cap_end_after_all_resolved(self) -> None:
+        """CAP END sent only after all caps are resolved."""
+        net = _net()
+        writer = MagicMock()
+        writer.is_closing.return_value = False
+        writer.drain = AsyncMock()
+        net._writer = writer
+        net._caps_pending = 1  # only server-time, no SASL
+
+        await net._handle(_msg(":server CAP * ACK :server-time"))
+        cap_end_calls = [c for c in writer.write.call_args_list
+                         if b"CAP END" in c.args[0]]
+        assert len(cap_end_calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_server_time_property(self) -> None:
+        net = _net()
+        assert net.server_time is False
+        net._server_time = True
+        assert net.server_time is True