test(spam): cover scheduler, config cache, and scoring edge cases

Raise diff/patch coverage above the Codecov gate (was 79.8%, target 80.29%)
by exercising the previously-untested spam paths:

- subscriber: TestSpamRescoreScheduler covers the disabled early-return, the
  enabled thread spawning + one sweep + clean stop, and the swallowed-error
  branch of the background re-scoring loop.
- handler: a contact-message scoring test covers the contact log branch.
- spam: get/reset_spam_config caching, the zero-weight combine path, the
  default-`now` path, and the null-sender reset in rescore_recent.

Patch coverage for the change is now ~97%.

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

tests/test_collector/test_handlers/test_message_spam.py

@@ -3,7 +3,10 @@
 from sqlalchemy import select
 
 import meshcore_hub.collector.handlers.message as handler_module
-from meshcore_hub.collector.handlers.message import handle_channel_message
+from meshcore_hub.collector.handlers.message import (
+    handle_channel_message,
+    handle_contact_message,
+)
 from meshcore_hub.collector.spam import SpamConfig
 from meshcore_hub.common.models import Message
 
@@ -82,6 +85,26 @@ class TestHandlerSpamEnabled:
         latest = max(msgs, key=lambda m: m.sender_timestamp)
         assert latest.spam_score >= ENABLED_CFG.score_threshold
 
+    def test_contact_message_scored_and_logged(
+        self, db_manager, db_session, monkeypatch
+    ):
+        """Contact messages are scored too (covers the contact log branch)."""
+        _enable_spam(monkeypatch)
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "text": "buy now",
+            "sender_name": "bob1",
+            "path_len": 6,
+            "path_hashes": ["AA", "BB", "CC", "DD", "EE", "FF"],
+        }
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+
+        msg = db_session.execute(select(Message)).scalar_one()
+        assert msg.message_type == "contact"
+        assert msg.sender_normalized == "bob"
+        assert msg.path_prefix == "AA,BB,CC"
+        assert msg.spam_score == 0.0
+
     def test_short_path_stores_null_prefix(self, db_manager, db_session, monkeypatch):
         _enable_spam(monkeypatch)
         handle_channel_message(

tests/test_collector/test_spam.py

@@ -5,6 +5,7 @@ fixture is configured for (SQLite by default, Postgres when
 ``TEST_DATABASE_BACKEND=postgres``).
 """
 
+from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 
 import pytest
@@ -12,8 +13,10 @@ import pytest
 from meshcore_hub.collector.spam import (
     SpamConfig,
     compute_path_prefix,
+    get_spam_config,
     normalize_sender,
     rescore_recent,
+    reset_spam_config,
     score_message,
 )
 from meshcore_hub.common.models import Message
@@ -296,3 +299,82 @@ class TestRescoreRecent:
 
         assert first >= 1
         assert second == 0  # nothing changes on a second pass
+
+
+class TestConfigCache:
+    def test_get_and_reset_spam_config(self):
+        """get_spam_config caches; reset clears so it rebuilds."""
+        reset_spam_config()
+        first = get_spam_config()
+        second = get_spam_config()
+        assert first is second  # cached instance
+        reset_spam_config()
+        third = get_spam_config()
+        assert third is not first  # rebuilt after reset
+        # Default env -> feature disabled.
+        assert third.enabled is False
+        reset_spam_config()
+
+
+class TestCombineEdgeCases:
+    def test_zero_weights_score_zero(self, db_session):
+        """When both weights are 0 the eligible-path branch returns 0.0."""
+        now = datetime.now(timezone.utc)
+        for i in range(6):
+            db_session.add(
+                _make_message(
+                    now - timedelta(seconds=5 + i),
+                    path_prefix="AA,BB,CC",
+                    sender_normalized="bob",
+                )
+            )
+        db_session.commit()
+
+        cfg = replace(CFG, weight_path=0.0, weight_name=0.0)
+        result = score_message(
+            db_session,
+            path_prefix="AA,BB,CC",
+            sender_normalized="bob",
+            path_len=6,
+            received_at=now,
+            cfg=cfg,
+        )
+        assert result.score == 0.0
+
+
+class TestRescoreEdgeCases:
+    def test_default_now(self, db_session):
+        """rescore_recent computes 'now' itself when not provided."""
+        now = datetime.now(timezone.utc)
+        for i in range(6):
+            db_session.add(
+                _make_message(
+                    now - timedelta(seconds=5 * i),
+                    path_prefix="AA,BB,CC",
+                    sender_normalized="bob",
+                )
+            )
+        db_session.commit()
+
+        updated = rescore_recent(db_session, CFG)  # now defaulted internally
+        db_session.commit()
+        assert updated >= 1
+
+    def test_null_sender_reset_to_zero(self, db_session):
+        """A recent row with no sender_normalized is reset to 0.0."""
+        now = datetime.now(timezone.utc)
+        m = _make_message(
+            now - timedelta(seconds=10),
+            path_prefix=None,
+            sender_normalized=None,
+            path_len=2,
+        )
+        m.spam_score = 0.5  # stale non-zero score
+        db_session.add(m)
+        db_session.commit()
+
+        updated = rescore_recent(db_session, CFG, now=now)
+        db_session.commit()
+        assert updated == 1
+        db_session.refresh(m)
+        assert m.spam_score == 0.0

tests/test_collector/test_subscriber.py

@@ -1290,3 +1290,85 @@ class TestChannelKeyRefresh:
         subscriber = Subscriber(mock_mqtt_client, db_manager)
 
         assert key_hex in subscriber._letsmesh_decoder._channel_keys
+
+
+class TestSpamRescoreScheduler:
+    """Tests for the background spam re-scoring sweep scheduler."""
+
+    @pytest.fixture
+    def mock_mqtt_client(self):
+        client = MagicMock()
+        client.topic_builder = MagicMock()
+        client.topic_builder.prefix = "meshcore"
+        return client
+
+    def test_disabled_does_not_start_thread(self, mock_mqtt_client, db_manager):
+        """With spam detection off (default), no sweep thread is created."""
+        sub = Subscriber(mock_mqtt_client, db_manager)
+        sub._start_spam_rescore_scheduler()
+        assert sub._spam_rescore_thread is None
+        # Stopping when never started is a no-op.
+        sub._stop_spam_rescore_scheduler()
+
+    def test_enabled_runs_sweep_and_stops(
+        self, mock_mqtt_client, db_manager, monkeypatch
+    ):
+        """Enabled scheduler spawns a thread, runs a sweep, and joins on stop."""
+        import threading
+
+        import meshcore_hub.collector.spam as spam_mod
+        from meshcore_hub.collector.spam import SpamConfig
+
+        cfg = SpamConfig(enabled=True, rescore_interval_seconds=1)
+        monkeypatch.setattr(spam_mod, "get_spam_config", lambda: cfg)
+        # Avoid the real 1s-per-tick sleep in the inner loop.
+        monkeypatch.setattr(
+            "meshcore_hub.collector.subscriber.time.sleep", lambda *_: None
+        )
+
+        sub = Subscriber(mock_mqtt_client, db_manager)
+        called = threading.Event()
+
+        def fake_rescore(session, sweep_cfg):
+            called.set()
+            sub._running = False  # break the loop after one sweep
+            return 1  # truthy -> exercises the "updated" log line
+
+        monkeypatch.setattr(spam_mod, "rescore_recent", fake_rescore)
+
+        sub._running = True
+        sub._start_spam_rescore_scheduler()
+        assert called.wait(timeout=5.0)
+        sub._stop_spam_rescore_scheduler()
+
+        assert sub._spam_rescore_thread is not None
+        assert not sub._spam_rescore_thread.is_alive()
+
+    def test_sweep_error_is_logged(self, mock_mqtt_client, db_manager, monkeypatch):
+        """An exception in the sweep is caught and logged, not propagated."""
+        import threading
+
+        import meshcore_hub.collector.spam as spam_mod
+        from meshcore_hub.collector.spam import SpamConfig
+
+        cfg = SpamConfig(enabled=True, rescore_interval_seconds=1)
+        monkeypatch.setattr(spam_mod, "get_spam_config", lambda: cfg)
+        monkeypatch.setattr(
+            "meshcore_hub.collector.subscriber.time.sleep", lambda *_: None
+        )
+
+        sub = Subscriber(mock_mqtt_client, db_manager)
+        called = threading.Event()
+
+        def boom(session, sweep_cfg):
+            sub._running = False  # break the loop before raising
+            called.set()
+            raise RuntimeError("sweep failed")
+
+        monkeypatch.setattr(spam_mod, "rescore_recent", boom)
+
+        sub._running = True
+        sub._start_spam_rescore_scheduler()
+        assert called.wait(timeout=5.0)
+        sub._stop_spam_rescore_scheduler()
+        assert not sub._spam_rescore_thread.is_alive()