data: resolve circular dependency of deamon.py (#653)

* data: resolve circular dependency of deamon.py

* address review comments

* address review comments

* address review comments

CLAUDE.md

@@ -15,7 +15,7 @@ Run linters for Python (`black`) and Ruby (`rufo`) to ensure consistent code for
 ## Project Structure & Module Organization
 The repository splits runtime and ingestion logic. `web/` holds the Sinatra dashboard (Ruby code in `lib/potato_mesh`, views in `views/`, static bundles in `public/`).
 
-`data/` hosts the Python Meshtastic ingestor plus migrations and CLI scripts. API fixtures and end-to-end harnesses live in `tests/`. Dockerfiles and compose files support containerized workflows.
+`data/` hosts the Python Meshtastic ingestor plus migrations and CLI scripts. The ingestor is structured as the `data/mesh_ingestor/` package with the following key modules: `daemon.py` (main loop), `handlers.py` (packet processing), `interfaces.py` (interface helpers), `config.py` (env-driven config), `events.py` (TypedDict event schemas), `provider.py` (Provider protocol), `node_identity.py` (canonical node ID utilities), `decode_payload.py` (CLI protobuf decoder), and the `providers/` subpackage (currently `meshtastic.py`). API contracts for all POST ingest routes are documented in `data/mesh_ingestor/CONTRACTS.md`. API fixtures and end-to-end harnesses live in `tests/`. Dockerfiles and compose files support containerized workflows.
 
 `matrix/` contains the Rust Matrix bridge; build with `cargo build --release` or `docker build -f matrix/Dockerfile .`, and keep bridge config under `matrix/Config.toml` when running locally.
 
@@ -39,7 +39,17 @@ Install dependencies with `cd app && flutter pub get`; format with `dart format
 ## Testing Guidelines
 Ruby specs run with `cd web && bundle exec rspec`, producing SimpleCov output in `coverage/`. Front-end behaviour is verified through Node’s test runner: `cd web && npm test` writes V8 coverage and JUnit XML under `reports/`.
 
-The ingestion layer is guarded by `pytest -q tests/test_mesh.py`; leave fixtures in `tests/` untouched so CI can replay them. New features should ship with matching specs and updated integration checks.
+The ingestion layer is tested with `pytest -q tests/`; leave fixtures in `tests/` untouched so CI can replay them. The suite includes both integration tests (`test_mesh.py`) and focused unit tests — `test_events_unit.py` (TypedDict schemas), `test_provider_unit.py` (Provider protocol conformance and `MeshtasticProvider`), `test_node_identity_unit.py` (canonical ID helpers), `test_daemon_unit.py`, `test_serialization_unit.py`, and `test_decode_payload.py`. New features should ship with matching specs and updated integration checks.
+
+## Adding a New Ingestor Provider
+The `data/mesh_ingestor/provider.py` module defines a `@runtime_checkable` `Provider` Protocol with five members: `name` (str), `subscribe()`, `connect(*, active_candidate)`, `extract_host_node_id(iface)`, and `node_snapshot_items(iface)`. To add a new backend (e.g. Reticulum, MeshCore):
+
+1. Create `data/mesh_ingestor/providers/<name>.py` with a class satisfying the Protocol.
+2. Register it in `data/mesh_ingestor/providers/__init__.py`.
+3. Pass an instance via `daemon.main(provider=...)` or make it the default in `main()`.
+4. Cover the provider with unit tests in `tests/test_provider_unit.py` — at minimum an `isinstance(..., Provider)` conformance check and any retry/error-handling paths.
+
+Consult `data/mesh_ingestor/CONTRACTS.md` for the canonical event shapes all providers must emit.
 
 ## Commit & Pull Request Guidelines
 Commits should stay imperative and reference issues the way history does (`Add chat log entries... (#408)`). Squash noisy work-in-progress commits before pushing. Pull requests need a concise summary, screenshots or curl traces for UI/API tweaks, and links to tracked issues. Paste the command output for the test suites you ran and mention configuration toggles (`API_TOKEN`, `PRIVATE`) reviewers must set.

data/mesh_ingestor/daemon.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import inspect
 import signal
 import threading
@@ -25,7 +26,6 @@ from pubsub import pub
 
 from . import config, handlers, ingestors, interfaces
 from .provider import Provider
-from .providers.meshtastic import MeshtasticProvider
 
 _RECEIVE_TOPICS = (
     "meshtastic.receive",
@@ -199,11 +199,6 @@ def _process_ingestor_heartbeat(iface, *, ingestor_announcement_sent: bool) -> b
     if heartbeat_sent and not ingestor_announcement_sent:
         return True
     return ingestor_announcement_sent
-    iface_cls = getattr(iface_obj, "__class__", None)
-    if iface_cls is None:
-        return False
-    module_name = getattr(iface_cls, "__module__", "") or ""
-    return "ble_interface" in module_name
 
 
 def _connected_state(candidate) -> bool | None:
@@ -245,10 +240,323 @@ def _connected_state(candidate) -> bool | None:
         return None
 
 
-def main(existing_interface=None, *, provider: Provider | None = None) -> None:
+# ---------------------------------------------------------------------------
+# Loop state container
+# ---------------------------------------------------------------------------
+
+
+@dataclasses.dataclass
+class _DaemonState:
+    """All mutable state for the :func:`main` daemon loop."""
+
+    provider: Provider
+    stop: threading.Event
+    configured_port: str | None
+    inactivity_reconnect_secs: float
+    energy_saving_enabled: bool
+    energy_online_secs: float
+    energy_sleep_secs: float
+    retry_delay: float
+    last_seen_packet_monotonic: float | None
+    active_candidate: str | None
+
+    iface: object = None
+    resolved_target: str | None = None
+    initial_snapshot_sent: bool = False
+    energy_session_deadline: float | None = None
+    iface_connected_at: float | None = None
+    last_inactivity_reconnect: float | None = None
+    ingestor_announcement_sent: bool = False
+    announced_target: bool = False
+
+
+# ---------------------------------------------------------------------------
+# Per-iteration helpers (each returns True when the caller should `continue`)
+# ---------------------------------------------------------------------------
+
+
+def _advance_retry_delay(current: float) -> float:
+    """Return the next exponential-backoff retry delay."""
+
+    if config._RECONNECT_MAX_DELAY_SECS <= 0:
+        return current
+    # `current == 0` on the very first call (bootstrap); seed from config.
+    next_delay = current * 2 if current else config._RECONNECT_INITIAL_DELAY_SECS
+    return min(next_delay, config._RECONNECT_MAX_DELAY_SECS)
+
+
+def _energy_sleep(state: _DaemonState, reason: str) -> None:
+    """Sleep for the configured energy-saving interval."""
+
+    if not state.energy_saving_enabled or state.energy_sleep_secs <= 0:
+        return
+    if config.DEBUG:
+        config._debug_log(
+            f"energy saving: {reason}; sleeping for {state.energy_sleep_secs:g}s"
+        )
+    state.stop.wait(state.energy_sleep_secs)
+
+
+def _try_connect(state: _DaemonState) -> bool:
+    """Attempt to establish the mesh interface.
+
+    Returns:
+        ``True`` when connected and the loop should proceed; ``False`` when
+        the connection failed and the caller should ``continue``.
+    """
+
+    try:
+        state.iface, state.resolved_target, state.active_candidate = (
+            state.provider.connect(active_candidate=state.active_candidate)
+        )
+        handlers.register_host_node_id(state.provider.extract_host_node_id(state.iface))
+        ingestors.set_ingestor_node_id(handlers.host_node_id())
+        state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
+        state.initial_snapshot_sent = False
+        if not state.announced_target and state.resolved_target:
+            config._debug_log(
+                "Using mesh interface",
+                context="daemon.interface",
+                severity="info",
+                target=state.resolved_target,
+            )
+            state.announced_target = True
+        if state.energy_saving_enabled and state.energy_online_secs > 0:
+            state.energy_session_deadline = time.monotonic() + state.energy_online_secs
+        else:
+            state.energy_session_deadline = None
+        state.iface_connected_at = time.monotonic()
+        # Seed the inactivity tracking from the connection time so a
+        # reconnect is given a full inactivity window even when the
+        # handler still reports the previous packet timestamp.
+        state.last_seen_packet_monotonic = state.iface_connected_at
+        state.last_inactivity_reconnect = None
+        return True
+    except interfaces.NoAvailableMeshInterface as exc:
+        config._debug_log(
+            "No mesh interface available",
+            context="daemon.interface",
+            severity="error",
+            error_message=str(exc),
+        )
+        _close_interface(state.iface)
+        raise SystemExit(1) from exc
+    except Exception as exc:
+        config._debug_log(
+            "Failed to create mesh interface",
+            context="daemon.interface",
+            severity="warn",
+            candidate=state.active_candidate or "auto",
+            error_class=exc.__class__.__name__,
+            error_message=str(exc),
+        )
+        if state.configured_port is None:
+            state.active_candidate = None
+            state.announced_target = False
+        state.stop.wait(state.retry_delay)
+        state.retry_delay = _advance_retry_delay(state.retry_delay)
+        return False
+
+
+def _check_energy_saving(state: _DaemonState) -> bool:
+    """Disconnect and sleep when energy-saving conditions are met.
+
+    Returns:
+        ``True`` when the interface was closed and the caller should
+        ``continue``; ``False`` otherwise.
+    """
+
+    if not state.energy_saving_enabled or state.iface is None:
+        return False
+
+    if (
+        state.energy_session_deadline is not None
+        and time.monotonic() >= state.energy_session_deadline
+    ):
+        reason = "disconnected after session"
+        log_msg = "Energy saving disconnect"
+    elif (
+        _is_ble_interface(state.iface)
+        and getattr(state.iface, "client", object()) is None
+    ):
+        reason = "BLE client disconnected"
+        log_msg = "Energy saving BLE disconnect"
+    else:
+        return False
+    config._debug_log(log_msg, context="daemon.energy", severity="info")
+    _close_interface(state.iface)
+    state.iface = None
+    state.announced_target = False
+    state.initial_snapshot_sent = False
+    state.energy_session_deadline = None
+    _energy_sleep(state, reason)
+    return True
+
+
+def _try_send_snapshot(state: _DaemonState) -> bool:
+    """Send the initial node snapshot via the provider.
+
+    Returns:
+        ``True`` when the snapshot succeeded (or no nodes exist yet); ``False``
+        when a hard error occurred and the caller should ``continue``.
+    """
+
+    try:
+        node_items = state.provider.node_snapshot_items(state.iface)
+        processed_any = False
+        for node_id, node in node_items:
+            processed_any = True
+            try:
+                handlers.upsert_node(node_id, node)
+            except Exception as exc:
+                config._debug_log(
+                    "Failed to update node snapshot",
+                    context="daemon.snapshot",
+                    severity="warn",
+                    node_id=node_id,
+                    error_class=exc.__class__.__name__,
+                    error_message=str(exc),
+                )
+                if config.DEBUG:
+                    config._debug_log(
+                        "Snapshot node payload",
+                        context="daemon.snapshot",
+                        node=node,
+                    )
+        if processed_any:
+            state.initial_snapshot_sent = True
+        return True
+    except Exception as exc:
+        config._debug_log(
+            "Snapshot refresh failed",
+            context="daemon.snapshot",
+            severity="warn",
+            error_class=exc.__class__.__name__,
+            error_message=str(exc),
+        )
+        _close_interface(state.iface)
+        state.iface = None
+        state.stop.wait(state.retry_delay)
+        state.retry_delay = _advance_retry_delay(state.retry_delay)
+        return False
+
+
+def _check_inactivity_reconnect(state: _DaemonState) -> bool:
+    """Reconnect when the interface has been silent for too long.
+
+    Returns:
+        ``True`` when a reconnect was triggered and the caller should
+        ``continue``; ``False`` otherwise.
+    """
+
+    if state.iface is None or state.inactivity_reconnect_secs <= 0:
+        return False
+
+    now = time.monotonic()
+    iface_activity = handlers.last_packet_monotonic()
+
+    if (
+        iface_activity is not None
+        and state.iface_connected_at is not None
+        and iface_activity < state.iface_connected_at
+    ):
+        iface_activity = state.iface_connected_at
+
+    if iface_activity is not None and (
+        state.last_seen_packet_monotonic is None
+        or iface_activity > state.last_seen_packet_monotonic
+    ):
+        state.last_seen_packet_monotonic = iface_activity
+        state.last_inactivity_reconnect = None
+
+    latest_activity = iface_activity
+    if latest_activity is None and state.iface_connected_at is not None:
+        latest_activity = state.iface_connected_at
+    if latest_activity is None:
+        latest_activity = now
+
+    inactivity_elapsed = now - latest_activity
+    believed_disconnected = (
+        _connected_state(getattr(state.iface, "isConnected", None)) is False
+    )
+
+    if (
+        not believed_disconnected
+        and inactivity_elapsed < state.inactivity_reconnect_secs
+    ):
+        return False
+
+    if (
+        state.last_inactivity_reconnect is not None
+        and now - state.last_inactivity_reconnect < state.inactivity_reconnect_secs
+    ):
+        return False
+
+    reason = (
+        "disconnected"
+        if believed_disconnected
+        else f"no data for {inactivity_elapsed:.0f}s"
+    )
+    config._debug_log(
+        "Mesh interface inactivity detected",
+        context="daemon.interface",
+        severity="warn",
+        reason=reason,
+    )
+    state.last_inactivity_reconnect = now
+    _close_interface(state.iface)
+    state.iface = None
+    state.announced_target = False
+    state.initial_snapshot_sent = False
+    state.energy_session_deadline = None
+    state.iface_connected_at = None
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Loop iteration helper
+# ---------------------------------------------------------------------------
+
+
+def _loop_iteration(state: _DaemonState) -> bool:
+    """Execute one pass of the daemon main loop.
+
+    Encapsulates the per-iteration ``continue`` decisions so that
+    :func:`main` stays within the allowed cognitive-complexity budget.
+
+    Returns:
+        ``True`` when the loop should start the next iteration immediately
+        (equivalent to a ``continue``); ``False`` when the full pass
+        completed and the caller should sleep before iterating again.
+    """
+
+    if state.iface is None and not _try_connect(state):
+        return True
+    if _check_energy_saving(state):
+        return True
+    if not state.initial_snapshot_sent and not _try_send_snapshot(state):
+        return True
+    if _check_inactivity_reconnect(state):
+        return True
+    state.ingestor_announcement_sent = _process_ingestor_heartbeat(
+        state.iface, ingestor_announcement_sent=state.ingestor_announcement_sent
+    )
+    state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
+    return False
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def main(*, provider: Provider | None = None) -> None:
     """Run the mesh ingestion daemon until interrupted."""
 
-    provider = provider or MeshtasticProvider()
+    if provider is None:
+        from .providers.meshtastic import MeshtasticProvider
+
+        provider = MeshtasticProvider()
 
     subscribed = provider.subscribe()
     if subscribed:
@@ -259,307 +567,72 @@ def main(existing_interface=None, *, provider: Provider | None = None) -> None:
             topics=subscribed,
         )
 
-    iface = existing_interface
-    resolved_target = None
-    retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
-
-    stop = threading.Event()
-    initial_snapshot_sent = False
-    energy_session_deadline = None
-    iface_connected_at: float | None = None
-    last_seen_packet_monotonic = handlers.last_packet_monotonic()
-    last_inactivity_reconnect: float | None = None
-    inactivity_reconnect_secs = max(
-        0.0, getattr(config, "_INACTIVITY_RECONNECT_SECS", 0.0)
+    state = _DaemonState(
+        provider=provider,
+        stop=threading.Event(),
+        configured_port=config.CONNECTION,
+        inactivity_reconnect_secs=max(
+            0.0, getattr(config, "_INACTIVITY_RECONNECT_SECS", 0.0)
+        ),
+        energy_saving_enabled=config.ENERGY_SAVING,
+        energy_online_secs=max(0.0, config._ENERGY_ONLINE_DURATION_SECS),
+        energy_sleep_secs=max(0.0, config._ENERGY_SLEEP_SECS),
+        retry_delay=max(0.0, config._RECONNECT_INITIAL_DELAY_SECS),
+        last_seen_packet_monotonic=handlers.last_packet_monotonic(),
+        active_candidate=config.CONNECTION,
     )
-    ingestor_announcement_sent = False
-
-    energy_saving_enabled = config.ENERGY_SAVING
-    energy_online_secs = max(0.0, config._ENERGY_ONLINE_DURATION_SECS)
-    energy_sleep_secs = max(0.0, config._ENERGY_SLEEP_SECS)
-
-    def _energy_sleep(reason: str) -> None:
-        if not energy_saving_enabled or energy_sleep_secs <= 0:
-            return
-        if config.DEBUG:
-            config._debug_log(
-                f"energy saving: {reason}; sleeping for {energy_sleep_secs:g}s"
-            )
-        stop.wait(energy_sleep_secs)
 
     def handle_sigterm(*_args) -> None:
-        stop.set()
+        state.stop.set()
 
     def handle_sigint(signum, frame) -> None:
-        if stop.is_set():
+        if state.stop.is_set():
             signal.default_int_handler(signum, frame)
             return
-        stop.set()
+        state.stop.set()
 
     if threading.current_thread() == threading.main_thread():
         signal.signal(signal.SIGINT, handle_sigint)
         signal.signal(signal.SIGTERM, handle_sigterm)
 
-    target = config.INSTANCE or "(no INSTANCE_DOMAIN configured)"
-    configured_port = config.CONNECTION
-    active_candidate = configured_port
-    announced_target = False
     config._debug_log(
         "Mesh daemon starting",
         context="daemon.main",
         severity="info",
-        target=target,
-        port=configured_port or "auto",
+        target=config.INSTANCE or "(no INSTANCE_DOMAIN configured)",
+        port=config.CONNECTION or "auto",
         channel=config.CHANNEL_INDEX,
     )
+
     try:
-        while not stop.is_set():
-            if iface is None:
-                try:
-                    iface, resolved_target, active_candidate = provider.connect(
-                        active_candidate=active_candidate
-                    )
-                    handlers.register_host_node_id(
-                        provider.extract_host_node_id(iface)
-                    )
-                    ingestors.set_ingestor_node_id(handlers.host_node_id())
-                    retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
-                    initial_snapshot_sent = False
-                    if not announced_target and resolved_target:
-                        config._debug_log(
-                            "Using mesh interface",
-                            context="daemon.interface",
-                            severity="info",
-                            target=resolved_target,
-                        )
-                        announced_target = True
-                    if energy_saving_enabled and energy_online_secs > 0:
-                        energy_session_deadline = time.monotonic() + energy_online_secs
-                    else:
-                        energy_session_deadline = None
-                    iface_connected_at = time.monotonic()
-                    # Seed the inactivity tracking from the connection time so a
-                    # reconnect is given a full inactivity window even when the
-                    # handler still reports the previous packet timestamp.
-                    last_seen_packet_monotonic = iface_connected_at
-                    last_inactivity_reconnect = None
-                except interfaces.NoAvailableMeshInterface as exc:
-                    config._debug_log(
-                        "No mesh interface available",
-                        context="daemon.interface",
-                        severity="error",
-                        error_message=str(exc),
-                    )
-                    _close_interface(iface)
-                    raise SystemExit(1) from exc
-                except Exception as exc:
-                    candidate_desc = active_candidate or "auto"
-                    config._debug_log(
-                        "Failed to create mesh interface",
-                        context="daemon.interface",
-                        severity="warn",
-                        candidate=candidate_desc,
-                        error_class=exc.__class__.__name__,
-                        error_message=str(exc),
-                    )
-                    if configured_port is None:
-                        active_candidate = None
-                        announced_target = False
-                    stop.wait(retry_delay)
-                    if config._RECONNECT_MAX_DELAY_SECS > 0:
-                        retry_delay = min(
-                            (
-                                retry_delay * 2
-                                if retry_delay
-                                else config._RECONNECT_INITIAL_DELAY_SECS
-                            ),
-                            config._RECONNECT_MAX_DELAY_SECS,
-                        )
-                    continue
-
-            if energy_saving_enabled and iface is not None:
-                if (
-                    energy_session_deadline is not None
-                    and time.monotonic() >= energy_session_deadline
-                ):
-                    config._debug_log(
-                        "Energy saving disconnect",
-                        context="daemon.energy",
-                        severity="info",
-                    )
-                    _close_interface(iface)
-                    iface = None
-                    announced_target = False
-                    initial_snapshot_sent = False
-                    energy_session_deadline = None
-                    _energy_sleep("disconnected after session")
-                    continue
-                if (
-                    _is_ble_interface(iface)
-                    and getattr(iface, "client", object()) is None
-                ):
-                    config._debug_log(
-                        "Energy saving BLE disconnect",
-                        context="daemon.energy",
-                        severity="info",
-                    )
-                    _close_interface(iface)
-                    iface = None
-                    announced_target = False
-                    initial_snapshot_sent = False
-                    energy_session_deadline = None
-                    _energy_sleep("BLE client disconnected")
-                    continue
-
-            if not initial_snapshot_sent:
-                try:
-                    node_items = list(provider.node_snapshot_items(iface))
-                    node_items = _node_items_snapshot(dict(node_items))
-                    if node_items is None:
-                        config._debug_log(
-                            "Skipping node snapshot due to concurrent modification",
-                            context="daemon.snapshot",
-                        )
-                    else:
-                        processed_snapshot_item = False
-                        for node_id, node in node_items:
-                            processed_snapshot_item = True
-                            try:
-                                handlers.upsert_node(node_id, node)
-                            except Exception as exc:
-                                config._debug_log(
-                                    "Failed to update node snapshot",
-                                    context="daemon.snapshot",
-                                    severity="warn",
-                                    node_id=node_id,
-                                    error_class=exc.__class__.__name__,
-                                    error_message=str(exc),
-                                )
-                                if config.DEBUG:
-                                    config._debug_log(
-                                        "Snapshot node payload",
-                                        context="daemon.snapshot",
-                                        node=node,
-                                    )
-                        if processed_snapshot_item:
-                            initial_snapshot_sent = True
-                except Exception as exc:
-                    config._debug_log(
-                        "Snapshot refresh failed",
-                        context="daemon.snapshot",
-                        severity="warn",
-                        error_class=exc.__class__.__name__,
-                        error_message=str(exc),
-                    )
-                    _close_interface(iface)
-                    iface = None
-                    stop.wait(retry_delay)
-                    if config._RECONNECT_MAX_DELAY_SECS > 0:
-                        retry_delay = min(
-                            (
-                                retry_delay * 2
-                                if retry_delay
-                                else config._RECONNECT_INITIAL_DELAY_SECS
-                            ),
-                            config._RECONNECT_MAX_DELAY_SECS,
-                        )
-                    continue
-
-            if iface is not None and inactivity_reconnect_secs > 0:
-                now_monotonic = time.monotonic()
-                iface_activity = handlers.last_packet_monotonic()
-                if (
-                    iface_activity is not None
-                    and iface_connected_at is not None
-                    and iface_activity < iface_connected_at
-                ):
-                    iface_activity = iface_connected_at
-                if iface_activity is not None and (
-                    last_seen_packet_monotonic is None
-                    or iface_activity > last_seen_packet_monotonic
-                ):
-                    last_seen_packet_monotonic = iface_activity
-                    last_inactivity_reconnect = None
-
-                latest_activity = iface_activity
-                if latest_activity is None and iface_connected_at is not None:
-                    latest_activity = iface_connected_at
-                if latest_activity is None:
-                    latest_activity = now_monotonic
-
-                inactivity_elapsed = now_monotonic - latest_activity
-
-                connected_attr = getattr(iface, "isConnected", None)
-                believed_disconnected = False
-                connected_state = _connected_state(connected_attr)
-                if connected_state is None:
-                    if callable(connected_attr):
-                        try:
-                            believed_disconnected = not bool(connected_attr())
-                        except Exception:
-                            believed_disconnected = False
-                    elif connected_attr is not None:
-                        try:
-                            believed_disconnected = not bool(connected_attr)
-                        except Exception:  # pragma: no cover - defensive guard
-                            believed_disconnected = False
-                else:
-                    believed_disconnected = not connected_state
-
-                should_reconnect = believed_disconnected or (
-                    inactivity_elapsed >= inactivity_reconnect_secs
-                )
-
-                if should_reconnect:
-                    if (
-                        last_inactivity_reconnect is None
-                        or now_monotonic - last_inactivity_reconnect
-                        >= inactivity_reconnect_secs
-                    ):
-                        reason = (
-                            "disconnected"
-                            if believed_disconnected
-                            else f"no data for {inactivity_elapsed:.0f}s"
-                        )
-                        config._debug_log(
-                            "Mesh interface inactivity detected",
-                            context="daemon.interface",
-                            severity="warn",
-                            reason=reason,
-                        )
-                        last_inactivity_reconnect = now_monotonic
-                        _close_interface(iface)
-                        iface = None
-                        announced_target = False
-                        initial_snapshot_sent = False
-                        energy_session_deadline = None
-                        iface_connected_at = None
-                        continue
-
-            ingestor_announcement_sent = _process_ingestor_heartbeat(
-                iface, ingestor_announcement_sent=ingestor_announcement_sent
-            )
-
-            retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
-            stop.wait(config.SNAPSHOT_SECS)
+        while not state.stop.is_set():
+            if not _loop_iteration(state):
+                state.stop.wait(config.SNAPSHOT_SECS)
     except KeyboardInterrupt:  # pragma: no cover - interactive only
         config._debug_log(
             "Received KeyboardInterrupt; shutting down",
             context="daemon.main",
             severity="info",
         )
-        stop.set()
+        state.stop.set()
     finally:
-        _close_interface(iface)
+        _close_interface(state.iface)
 
 
 __all__ = [
     "_RECEIVE_TOPICS",
-    "_event_wait_allows_default_timeout",
-    "_node_items_snapshot",
-    "_subscribe_receive_topics",
-    "_is_ble_interface",
-    "_process_ingestor_heartbeat",
+    "_advance_retry_delay",
+    "_loop_iteration",
+    "_check_energy_saving",
+    "_check_inactivity_reconnect",
     "_connected_state",
+    "_energy_sleep",
+    "_event_wait_allows_default_timeout",
+    "_is_ble_interface",
+    "_node_items_snapshot",
+    "_process_ingestor_heartbeat",
+    "_subscribe_receive_topics",
+    "_try_connect",
+    "_try_send_snapshot",
     "main",
 ]

data/mesh_ingestor/events.py

@@ -27,10 +27,13 @@ from __future__ import annotations
 from typing import NotRequired, TypedDict
 
 
-class MessageEvent(TypedDict, total=False):
+class _MessageEventRequired(TypedDict):
     id: int
     rx_time: int
     rx_iso: str
+
+
+class MessageEvent(_MessageEventRequired, total=False):
     from_id: object
     to_id: object
     channel: int
@@ -48,15 +51,18 @@ class MessageEvent(TypedDict, total=False):
     modem_preset: str
 
 
-class PositionEvent(TypedDict, total=False):
+class _PositionEventRequired(TypedDict):
     id: int
+    rx_time: int
+    rx_iso: str
+
+
+class PositionEvent(_PositionEventRequired, total=False):
     node_id: str
     node_num: int | None
     num: int | None
     from_id: str | None
     to_id: object
-    rx_time: int
-    rx_iso: str
     latitude: float | None
     longitude: float | None
     altitude: float | None
@@ -78,14 +84,17 @@ class PositionEvent(TypedDict, total=False):
     modem_preset: str
 
 
-class TelemetryEvent(TypedDict, total=False):
+class _TelemetryEventRequired(TypedDict):
     id: int
+    rx_time: int
+    rx_iso: str
+
+
+class TelemetryEvent(_TelemetryEventRequired, total=False):
     node_id: str | None
     node_num: int | None
     from_id: object
     to_id: object
-    rx_time: int
-    rx_iso: str
     telemetry_time: int | None
     channel: int
     portnum: str | None
@@ -102,20 +111,26 @@ class TelemetryEvent(TypedDict, total=False):
     # evolves over time.
 
 
-class NeighborEntry(TypedDict, total=False):
+class _NeighborEntryRequired(TypedDict):
+    rx_time: int
+    rx_iso: str
+
+
+class NeighborEntry(_NeighborEntryRequired, total=False):
     neighbor_id: str
     neighbor_num: int | None
     snr: float | None
+
+
+class _NeighborsSnapshotRequired(TypedDict):
+    node_id: str
     rx_time: int
     rx_iso: str
 
 
-class NeighborsSnapshot(TypedDict, total=False):
-    node_id: str
+class NeighborsSnapshot(_NeighborsSnapshotRequired, total=False):
     node_num: int | None
     neighbors: list[NeighborEntry]
-    rx_time: int
-    rx_iso: str
     node_broadcast_interval_secs: int | None
     last_sent_by_id: str | None
     ingestor: str | None
@@ -123,14 +138,17 @@ class NeighborsSnapshot(TypedDict, total=False):
     modem_preset: str
 
 
-class TraceEvent(TypedDict, total=False):
+class _TraceEventRequired(TypedDict):
+    hops: list[int]
+    rx_time: int
+    rx_iso: str
+
+
+class TraceEvent(_TraceEventRequired, total=False):
     id: int | None
     request_id: int | None
     src: int | None
     dest: int | None
-    rx_time: int
-    rx_iso: str
-    hops: list[int]
     rssi: int | None
     snr: float | None
     elapsed_ms: int | None
@@ -161,4 +179,3 @@ __all__ = [
     "TelemetryEvent",
     "TraceEvent",
 ]
-

data/mesh_ingestor/node_identity.py

@@ -25,8 +25,7 @@ from __future__ import annotations
 
 from typing import Final
 
-
-_CANONICAL_PREFIX: Final[str] = "!"
+CANONICAL_PREFIX: Final[str] = "!"
 
 
 def canonical_node_id(value: object) -> str | None:
@@ -48,7 +47,7 @@ def canonical_node_id(value: object) -> str | None:
             return None
         if num < 0:
             return None
-        return f"{_CANONICAL_PREFIX}{num & 0xFFFFFFFF:08x}"
+        return f"{CANONICAL_PREFIX}{num & 0xFFFFFFFF:08x}"
     if not isinstance(value, str):
         return None
 
@@ -59,13 +58,13 @@ def canonical_node_id(value: object) -> str | None:
         # Meshtastic special destinations like "^all" are not node ids; callers
         # that already accept them should keep passing them through unchanged.
         return trimmed
-    if trimmed.startswith(_CANONICAL_PREFIX):
+    if trimmed.startswith(CANONICAL_PREFIX):
         body = trimmed[1:]
     elif trimmed.lower().startswith("0x"):
         body = trimmed[2:]
     elif trimmed.isdigit():
         try:
-            return f"{_CANONICAL_PREFIX}{int(trimmed, 10) & 0xFFFFFFFF:08x}"
+            return f"{CANONICAL_PREFIX}{int(trimmed, 10) & 0xFFFFFFFF:08x}"
         except ValueError:
             return None
     else:
@@ -74,7 +73,7 @@ def canonical_node_id(value: object) -> str | None:
     if not body:
         return None
     try:
-        return f"{_CANONICAL_PREFIX}{int(body, 16) & 0xFFFFFFFF:08x}"
+        return f"{CANONICAL_PREFIX}{int(body, 16) & 0xFFFFFFFF:08x}"
     except ValueError:
         return None
 
@@ -96,7 +95,7 @@ def node_num_from_id(node_id: object) -> int | None:
     trimmed = node_id.strip()
     if not trimmed:
         return None
-    if trimmed.startswith(_CANONICAL_PREFIX):
+    if trimmed.startswith(CANONICAL_PREFIX):
         trimmed = trimmed[1:]
     if trimmed.lower().startswith("0x"):
         trimmed = trimmed[2:]
@@ -110,7 +109,7 @@ def node_num_from_id(node_id: object) -> int | None:
 
 
 __all__ = [
+    "CANONICAL_PREFIX",
     "canonical_node_id",
     "node_num_from_id",
 ]
-

data/mesh_ingestor/provider.py

@@ -23,7 +23,8 @@ from __future__ import annotations
 
 import enum
 from collections.abc import Iterable
-from typing import Protocol
+from typing import Protocol, runtime_checkable
+
 
 class ProviderCapability(enum.Flag):
     """Feature flags describing what a provider can supply."""
@@ -33,11 +34,11 @@ class ProviderCapability(enum.Flag):
     HEARTBEATS = enum.auto()
 
 
+@runtime_checkable
 class Provider(Protocol):
     """Abstract source of mesh observations."""
 
     name: str
-    capabilities: ProviderCapability
 
     def subscribe(self) -> list[str]:
         """Subscribe to any async receive callbacks and return topic names."""
@@ -62,4 +63,3 @@ __all__ = [
     "Provider",
     "ProviderCapability",
 ]
-

data/mesh_ingestor/providers/__init__.py

@@ -23,4 +23,3 @@ from __future__ import annotations
 from .meshtastic import MeshtasticProvider
 
 __all__ = ["MeshtasticProvider"]
-

data/mesh_ingestor/providers/meshtastic.py

@@ -16,17 +16,17 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterable
+import time
 
-from .. import interfaces
-from ..provider import ProviderCapability
+from pubsub import pub
+
+from .. import config, daemon as _daemon, handlers, interfaces
 
 
 class MeshtasticProvider:
     """Meshtastic ingestion provider (current default)."""
 
     name = "meshtastic"
-    capabilities = ProviderCapability.NODE_SNAPSHOT | ProviderCapability.HEARTBEATS
 
     def __init__(self):
         self._subscribed: list[str] = []
@@ -37,14 +37,15 @@ class MeshtasticProvider:
         if self._subscribed:
             return list(self._subscribed)
 
-        # Delegate to the historical subscription helper in `daemon.py` so unit
-        # tests can monkeypatch the subscription mechanism via `daemon.pub`.
-        from .. import daemon as _daemon  # local import avoids module cycles
-
-        topics = _daemon._subscribe_receive_topics()
-
-        self._subscribed = topics
-        return list(topics)
+        subscribed = []
+        for topic in _daemon._RECEIVE_TOPICS:
+            try:
+                pub.subscribe(handlers.on_receive, topic)
+                subscribed.append(topic)
+            except Exception as exc:  # pragma: no cover
+                config._debug_log(f"failed to subscribe to {topic!r}: {exc}")
+        self._subscribed = subscribed
+        return list(subscribed)
 
     def connect(
         self, *, active_candidate: str | None
@@ -56,7 +57,9 @@ class MeshtasticProvider:
         next_candidate = active_candidate
 
         if active_candidate:
-            iface, resolved_target = interfaces._create_serial_interface(active_candidate)
+            iface, resolved_target = interfaces._create_serial_interface(
+                active_candidate
+            )
         else:
             iface, resolved_target = interfaces._create_default_interface()
             next_candidate = resolved_target
@@ -69,16 +72,20 @@ class MeshtasticProvider:
     def extract_host_node_id(self, iface: object) -> str | None:
         return interfaces._extract_host_node_id(iface)
 
-    def node_snapshot_items(self, iface: object) -> Iterable[tuple[str, object]]:
+    def node_snapshot_items(self, iface: object) -> list[tuple[str, object]]:
         nodes = getattr(iface, "nodes", {}) or {}
-        items_callable = getattr(nodes, "items", None)
-        if callable(items_callable):
-            return list(items_callable())
-        if hasattr(nodes, "__iter__") and hasattr(nodes, "__getitem__"):
-            keys = list(nodes)
-            return [(key, nodes[key]) for key in keys]
+        for _ in range(3):
+            try:
+                return list(nodes.items())
+            except RuntimeError as err:
+                if "dictionary changed size during iteration" not in str(err):
+                    raise
+                time.sleep(0)
+        config._debug_log(
+            "Skipping node snapshot due to concurrent modification",
+            context="meshtastic.snapshot",
+        )
         return []
 
 
 __all__ = ["MeshtasticProvider"]
-

tests/test_daemon_unit.py

@@ -435,3 +435,391 @@ def test_main_inactivity_reconnect(monkeypatch):
 
     daemon.main()
     assert any(event.is_set() for event in FakeEvent.instances)
+
+
+# ---------------------------------------------------------------------------
+# Helper: build a minimal _DaemonState for unit tests
+# ---------------------------------------------------------------------------
+
+
+def _make_state(**overrides):
+    """Return a :class:`daemon._DaemonState` with sensible defaults.
+
+    Any keyword argument is forwarded as a field override via ``setattr``
+    after construction, so callers only need to supply fields under test.
+    """
+    state = daemon._DaemonState(
+        provider=None,  # type: ignore[arg-type]
+        stop=FakeEvent(),  # type: ignore[arg-type]
+        configured_port=None,
+        inactivity_reconnect_secs=0.0,
+        energy_saving_enabled=False,
+        energy_online_secs=0.0,
+        energy_sleep_secs=0.0,
+        retry_delay=0.0,
+        last_seen_packet_monotonic=None,
+        active_candidate=None,
+    )
+    for key, val in overrides.items():
+        setattr(state, key, val)
+    return state
+
+
+# ---------------------------------------------------------------------------
+# _advance_retry_delay
+# ---------------------------------------------------------------------------
+
+
+def test_advance_retry_delay_disabled(monkeypatch):
+    """Returns current delay unchanged when the max is zero."""
+    monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0)
+    assert daemon._advance_retry_delay(5.0) == 5.0
+
+
+def test_advance_retry_delay_bootstrap(monkeypatch):
+    """Seeds from initial config when current delay is zero (first call)."""
+    monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 60.0)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 3.0)
+    assert daemon._advance_retry_delay(0.0) == 3.0
+
+
+def test_advance_retry_delay_doubles_and_caps(monkeypatch):
+    """Doubles current delay and caps at the configured maximum."""
+    monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 10.0)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 1.0)
+    assert daemon._advance_retry_delay(3.0) == 6.0
+    assert daemon._advance_retry_delay(7.0) == 10.0
+
+
+# ---------------------------------------------------------------------------
+# _energy_sleep
+# ---------------------------------------------------------------------------
+
+
+def test_energy_sleep_no_op_when_disabled():
+    """No wait issued when energy saving is disabled."""
+    state = _make_state(energy_saving_enabled=False, energy_sleep_secs=1.0)
+    daemon._energy_sleep(state, "reason")
+    assert not state.stop.wait_calls
+
+
+def test_energy_sleep_no_op_when_zero_secs():
+    """No wait issued when sleep duration is zero."""
+    state = _make_state(energy_saving_enabled=True, energy_sleep_secs=0.0)
+    daemon._energy_sleep(state, "reason")
+    assert not state.stop.wait_calls
+
+
+def test_energy_sleep_emits_debug_log(monkeypatch):
+    """Debug log is emitted when DEBUG is enabled."""
+    state = _make_state(energy_saving_enabled=True, energy_sleep_secs=2.0)
+    logged = []
+    monkeypatch.setattr(daemon.config, "DEBUG", True)
+    monkeypatch.setattr(
+        daemon.config, "_debug_log", lambda msg, **_kw: logged.append(msg)
+    )
+    daemon._energy_sleep(state, "wake up")
+    assert any("wake up" in m for m in logged)
+    assert state.stop.wait_calls == [2.0]
+
+
+def test_energy_sleep_waits_when_debug_off(monkeypatch):
+    """Wait is issued for the configured duration when DEBUG is off."""
+    state = _make_state(energy_saving_enabled=True, energy_sleep_secs=1.5)
+    monkeypatch.setattr(daemon.config, "DEBUG", False)
+    daemon._energy_sleep(state, "reason")
+    assert state.stop.wait_calls == [1.5]
+
+
+# ---------------------------------------------------------------------------
+# _try_connect
+# ---------------------------------------------------------------------------
+
+
+def test_try_connect_no_available_interface_raises_system_exit(monkeypatch):
+    """NoAvailableMeshInterface propagates as SystemExit(1)."""
+
+    class _NoIface:
+        def connect(self, *, active_candidate):
+            raise daemon.interfaces.NoAvailableMeshInterface("none")
+
+        def extract_host_node_id(self, iface):
+            return None
+
+    state = _make_state(active_candidate="serial0", configured_port="serial0")
+    state.provider = _NoIface()  # type: ignore[assignment]
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+    with pytest.raises(SystemExit):
+        daemon._try_connect(state)
+
+
+def test_try_connect_generic_failure_resets_candidate(monkeypatch):
+    """Connect failure in auto-detect mode clears the active candidate."""
+
+    class _FailProvider:
+        def connect(self, *, active_candidate):
+            raise OSError("device busy")
+
+        def extract_host_node_id(self, iface):
+            return None
+
+    state = _make_state(active_candidate="serial0", configured_port=None)
+    state.provider = _FailProvider()  # type: ignore[assignment]
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0)
+
+    result = daemon._try_connect(state)
+    assert result is False
+    assert state.active_candidate is None
+    assert state.announced_target is False
+
+
+def test_try_connect_sets_energy_session_deadline(monkeypatch):
+    """Energy-saving deadline is assigned when online duration is positive."""
+
+    class _OkProvider:
+        def connect(self, *, active_candidate):
+            return DummyInterface(), active_candidate, active_candidate
+
+        def extract_host_node_id(self, iface):
+            return "!host"
+
+    state = _make_state(
+        active_candidate="serial0",
+        configured_port="serial0",
+        energy_saving_enabled=True,
+        energy_online_secs=30.0,
+    )
+    state.provider = _OkProvider()  # type: ignore[assignment]
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0)
+    monkeypatch.setattr(
+        daemon.handlers, "register_host_node_id", lambda *_a, **_k: None
+    )
+    monkeypatch.setattr(daemon.handlers, "host_node_id", lambda: "!host")
+    monkeypatch.setattr(
+        daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None
+    )
+
+    result = daemon._try_connect(state)
+    assert result is True
+    assert state.energy_session_deadline is not None
+
+
+# ---------------------------------------------------------------------------
+# _check_energy_saving
+# ---------------------------------------------------------------------------
+
+
+def test_check_energy_saving_session_expired(monkeypatch):
+    """Iface is closed and True returned when the session deadline has passed."""
+    state = _make_state(energy_saving_enabled=True)
+    state.iface = DummyInterface()
+    state.energy_session_deadline = 0.0
+    monkeypatch.setattr(daemon.time, "monotonic", lambda: 1.0)
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+
+    result = daemon._check_energy_saving(state)
+    assert result is True
+    assert state.iface is None
+    assert state.energy_session_deadline is None
+
+
+def test_check_energy_saving_ble_client_disconnected(monkeypatch):
+    """Iface is closed and True returned when the BLE client reference is gone."""
+    state = _make_state(energy_saving_enabled=True)
+    state.iface = DummyInterface(client_present=False)
+    state.energy_session_deadline = None
+    monkeypatch.setattr(daemon, "_is_ble_interface", lambda _: True)
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+
+    result = daemon._check_energy_saving(state)
+    assert result is True
+    assert state.iface is None
+
+
+# ---------------------------------------------------------------------------
+# _try_send_snapshot
+# ---------------------------------------------------------------------------
+
+
+def test_try_send_snapshot_empty_nodes():
+    """Returns True without setting initial_snapshot_sent when no nodes exist."""
+
+    class _EmptyProvider:
+        def node_snapshot_items(self, iface):
+            return []
+
+    state = _make_state()
+    state.iface = DummyInterface(nodes={})
+    state.provider = _EmptyProvider()  # type: ignore[assignment]
+
+    result = daemon._try_send_snapshot(state)
+    assert result is True
+    assert state.initial_snapshot_sent is False
+
+
+def test_try_send_snapshot_upsert_failure_is_non_fatal(monkeypatch):
+    """Upsert errors are logged but do not abort the snapshot pass."""
+
+    class _OneNodeProvider:
+        def node_snapshot_items(self, iface):
+            return [("!node1", {"id": 1})]
+
+    def _raise(*_a, **_k):
+        raise ValueError("bad node")
+
+    state = _make_state()
+    state.iface = DummyInterface()
+    state.provider = _OneNodeProvider()  # type: ignore[assignment]
+    logged = []
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw))
+    monkeypatch.setattr(daemon.config, "DEBUG", False)
+    monkeypatch.setattr(daemon.handlers, "upsert_node", _raise)
+
+    result = daemon._try_send_snapshot(state)
+    assert result is True
+    assert state.initial_snapshot_sent is True
+    assert any(c.get("context") == "daemon.snapshot" for c in logged)
+
+
+def test_try_send_snapshot_upsert_failure_debug_payload(monkeypatch):
+    """The node payload is logged when DEBUG is enabled and upsert fails."""
+
+    class _OneNodeProvider:
+        def node_snapshot_items(self, iface):
+            return [("!node1", {"id": 1})]
+
+    def _raise(*_a, **_k):
+        raise ValueError("bad")
+
+    state = _make_state()
+    state.iface = DummyInterface()
+    state.provider = _OneNodeProvider()  # type: ignore[assignment]
+    logged = []
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *a, **kw: logged.append(kw))
+    monkeypatch.setattr(daemon.config, "DEBUG", True)
+    monkeypatch.setattr(daemon.handlers, "upsert_node", _raise)
+
+    daemon._try_send_snapshot(state)
+    assert any("node" in c for c in logged)
+
+
+def test_try_send_snapshot_outer_exception_resets_iface(monkeypatch):
+    """An exception from node_snapshot_items resets the interface and returns False."""
+
+    class _BrokenProvider:
+        def node_snapshot_items(self, iface):
+            raise RuntimeError("boom")
+
+    state = _make_state()
+    state.iface = DummyInterface()
+    state.provider = _BrokenProvider()  # type: ignore[assignment]
+    monkeypatch.setattr(daemon.config, "_debug_log", lambda *_a, **_k: None)
+    monkeypatch.setattr(daemon.config, "_RECONNECT_MAX_DELAY_SECS", 0)
+
+    result = daemon._try_send_snapshot(state)
+    assert result is False
+    assert state.iface is None
+
+
+# ---------------------------------------------------------------------------
+# _check_inactivity_reconnect (additional branches)
+# ---------------------------------------------------------------------------
+
+
+def test_check_inactivity_reconnect_throttles_rapid_reconnects(monkeypatch):
+    """A reconnect within the inactivity window is suppressed."""
+    state = _make_state(inactivity_reconnect_secs=60.0)
+    state.iface = DummyInterface(is_connected=False)
+    state.iface_connected_at = 0.0
+    state.last_inactivity_reconnect = 1.0  # recent
+
+    monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0)
+    monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None)
+
+    assert daemon._check_inactivity_reconnect(state) is False
+
+
+def test_check_inactivity_reconnect_uses_connected_at_when_no_packets(monkeypatch):
+    """Uses iface_connected_at as the activity baseline when no packets seen."""
+    state = _make_state(inactivity_reconnect_secs=60.0)
+    state.iface = DummyInterface(is_connected=True)
+    state.iface_connected_at = 5.0
+    state.last_inactivity_reconnect = None
+
+    monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0)
+    monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None)
+
+    # 10.0 - 5.0 = 5.0 < 60.0 → not triggered
+    assert daemon._check_inactivity_reconnect(state) is False
+
+
+def test_check_inactivity_reconnect_uses_now_when_no_baseline(monkeypatch):
+    """Falls back to current time when neither packets nor connected_at is set."""
+    state = _make_state(inactivity_reconnect_secs=60.0)
+    state.iface = DummyInterface(is_connected=True)
+    state.iface_connected_at = None
+    state.last_inactivity_reconnect = None
+
+    monkeypatch.setattr(daemon.time, "monotonic", lambda: 10.0)
+    monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None)
+
+    # latest_activity = now(10.0); inactivity_elapsed = 0.0 < 60.0 → not triggered
+    assert daemon._check_inactivity_reconnect(state) is False
+
+
+# ---------------------------------------------------------------------------
+# _loop_iteration
+# ---------------------------------------------------------------------------
+
+
+def test_loop_iteration_connect_fails_returns_true(monkeypatch):
+    """Returns True (continue) when iface is absent and connect fails."""
+    state = _make_state()
+    state.iface = None
+    monkeypatch.setattr(daemon, "_try_connect", lambda s: False)
+    assert daemon._loop_iteration(state) is True
+
+
+def test_loop_iteration_energy_saving_triggers_returns_true(monkeypatch):
+    """Returns True (continue) when energy saving disconnects the interface."""
+    state = _make_state()
+    state.iface = object()
+    monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: True)
+    assert daemon._loop_iteration(state) is True
+
+
+def test_loop_iteration_snapshot_fails_returns_true(monkeypatch):
+    """Returns True (continue) when the initial snapshot fails."""
+    state = _make_state()
+    state.iface = object()
+    state.initial_snapshot_sent = False
+    monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False)
+    monkeypatch.setattr(daemon, "_try_send_snapshot", lambda s: False)
+    assert daemon._loop_iteration(state) is True
+
+
+def test_loop_iteration_inactivity_triggers_returns_true(monkeypatch):
+    """Returns True (continue) when inactivity reconnect fires."""
+    state = _make_state()
+    state.iface = object()
+    state.initial_snapshot_sent = True
+    monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False)
+    monkeypatch.setattr(daemon, "_check_inactivity_reconnect", lambda s: True)
+    assert daemon._loop_iteration(state) is True
+
+
+def test_loop_iteration_full_pass_returns_false(monkeypatch):
+    """Returns False (sleep) after a complete iteration with no early exits."""
+    state = _make_state()
+    state.iface = object()
+    state.initial_snapshot_sent = True
+    monkeypatch.setattr(daemon, "_check_energy_saving", lambda s: False)
+    monkeypatch.setattr(daemon, "_check_inactivity_reconnect", lambda s: False)
+    monkeypatch.setattr(
+        daemon, "_process_ingestor_heartbeat", lambda iface, **_kw: False
+    )
+    monkeypatch.setattr(daemon.config, "_RECONNECT_INITIAL_DELAY_SECS", 0)
+    assert daemon._loop_iteration(state) is False

tests/test_events_unit.py

@@ -0,0 +1,232 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for :mod:`data.mesh_ingestor.events`."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from data.mesh_ingestor.events import (  # noqa: E402 - path setup
+    IngestorHeartbeat,
+    MessageEvent,
+    NeighborEntry,
+    NeighborsSnapshot,
+    PositionEvent,
+    TelemetryEvent,
+    TraceEvent,
+)
+
+
+def test_message_event_schema():
+    assert MessageEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"})
+    assert "text" in MessageEvent.__optional_keys__
+    assert "from_id" in MessageEvent.__optional_keys__
+    assert "snr" in MessageEvent.__optional_keys__
+    assert "rssi" in MessageEvent.__optional_keys__
+
+
+def test_message_event_requires_id_rx_time_rx_iso():
+    event: MessageEvent = {
+        "id": 1,
+        "rx_time": 1700000000,
+        "rx_iso": "2023-11-14T00:00:00Z",
+    }
+    assert event["id"] == 1
+    assert event["rx_time"] == 1700000000
+    assert event["rx_iso"] == "2023-11-14T00:00:00Z"
+
+
+def test_message_event_accepts_optional_fields():
+    event: MessageEvent = {
+        "id": 2,
+        "rx_time": 1700000001,
+        "rx_iso": "2023-11-14T00:00:01Z",
+        "text": "hello",
+        "from_id": "!aabbccdd",
+        "snr": 4.5,
+        "rssi": -90,
+    }
+    assert event["text"] == "hello"
+    assert event["snr"] == pytest.approx(4.5)
+
+
+def test_position_event_schema():
+    assert PositionEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"})
+    assert "latitude" in PositionEvent.__optional_keys__
+    assert "longitude" in PositionEvent.__optional_keys__
+    assert "node_id" in PositionEvent.__optional_keys__
+
+
+def test_position_event_required_fields():
+    event: PositionEvent = {
+        "id": 10,
+        "rx_time": 1700000002,
+        "rx_iso": "2023-11-14T00:00:02Z",
+    }
+    assert event["id"] == 10
+
+
+def test_position_event_optional_fields():
+    event: PositionEvent = {
+        "id": 11,
+        "rx_time": 1700000003,
+        "rx_iso": "2023-11-14T00:00:03Z",
+        "latitude": 37.7749,
+        "longitude": -122.4194,
+        "altitude": 10.0,
+        "node_id": "!aabbccdd",
+    }
+    assert event["latitude"] == pytest.approx(37.7749)
+
+
+def test_telemetry_event_schema():
+    assert TelemetryEvent.__required_keys__ == frozenset({"id", "rx_time", "rx_iso"})
+    assert "payload_b64" in TelemetryEvent.__optional_keys__
+    assert "snr" in TelemetryEvent.__optional_keys__
+
+
+def test_telemetry_event_required_fields():
+    event: TelemetryEvent = {
+        "id": 20,
+        "rx_time": 1700000004,
+        "rx_iso": "2023-11-14T00:00:04Z",
+    }
+    assert event["id"] == 20
+
+
+def test_telemetry_event_optional_fields():
+    event: TelemetryEvent = {
+        "id": 21,
+        "rx_time": 1700000005,
+        "rx_iso": "2023-11-14T00:00:05Z",
+        "channel": 0,
+        "payload_b64": "AAEC",
+        "snr": 3.0,
+    }
+    assert event["payload_b64"] == "AAEC"
+
+
+def test_neighbor_entry_schema():
+    assert NeighborEntry.__required_keys__ == frozenset({"rx_time", "rx_iso"})
+    assert "neighbor_id" in NeighborEntry.__optional_keys__
+    assert "snr" in NeighborEntry.__optional_keys__
+
+
+def test_neighbor_entry_required_fields():
+    entry: NeighborEntry = {"rx_time": 1700000006, "rx_iso": "2023-11-14T00:00:06Z"}
+    assert entry["rx_time"] == 1700000006
+
+
+def test_neighbor_entry_optional_fields():
+    entry: NeighborEntry = {
+        "rx_time": 1700000007,
+        "rx_iso": "2023-11-14T00:00:07Z",
+        "neighbor_id": "!11223344",
+        "snr": 6.0,
+    }
+    assert entry["neighbor_id"] == "!11223344"
+
+
+def test_neighbors_snapshot_schema():
+    assert NeighborsSnapshot.__required_keys__ == frozenset(
+        {"node_id", "rx_time", "rx_iso"}
+    )
+    assert "neighbors" in NeighborsSnapshot.__optional_keys__
+    assert "node_broadcast_interval_secs" in NeighborsSnapshot.__optional_keys__
+
+
+def test_neighbors_snapshot_required_fields():
+    snap: NeighborsSnapshot = {
+        "node_id": "!aabbccdd",
+        "rx_time": 1700000008,
+        "rx_iso": "2023-11-14T00:00:08Z",
+    }
+    assert snap["node_id"] == "!aabbccdd"
+
+
+def test_neighbors_snapshot_optional_fields():
+    snap: NeighborsSnapshot = {
+        "node_id": "!aabbccdd",
+        "rx_time": 1700000009,
+        "rx_iso": "2023-11-14T00:00:09Z",
+        "neighbors": [],
+        "node_broadcast_interval_secs": 900,
+    }
+    assert snap["node_broadcast_interval_secs"] == 900
+
+
+def test_trace_event_schema():
+    assert TraceEvent.__required_keys__ == frozenset({"hops", "rx_time", "rx_iso"})
+    assert "elapsed_ms" in TraceEvent.__optional_keys__
+    assert "snr" in TraceEvent.__optional_keys__
+
+
+def test_trace_event_required_fields():
+    event: TraceEvent = {
+        "hops": [1, 2, 3],
+        "rx_time": 1700000010,
+        "rx_iso": "2023-11-14T00:00:10Z",
+    }
+    assert event["hops"] == [1, 2, 3]
+
+
+def test_trace_event_optional_fields():
+    event: TraceEvent = {
+        "hops": [4, 5],
+        "rx_time": 1700000011,
+        "rx_iso": "2023-11-14T00:00:11Z",
+        "elapsed_ms": 42,
+        "snr": 2.5,
+    }
+    assert event["elapsed_ms"] == 42
+
+
+def test_ingestor_heartbeat_schema():
+    # IngestorHeartbeat uses total=True with NotRequired fields. Under
+    # `from __future__ import annotations` the TypedDict metaclass cannot
+    # evaluate the annotation strings at class creation time, so
+    # NotRequired keys appear in __required_keys__ rather than
+    # __optional_keys__. Verify the four always-present keys are included.
+    always_required = {"node_id", "start_time", "last_seen_time", "version"}
+    assert always_required <= IngestorHeartbeat.__required_keys__
+
+
+def test_ingestor_heartbeat_all_fields():
+    hb: IngestorHeartbeat = {
+        "node_id": "!aabbccdd",
+        "start_time": 1700000000,
+        "last_seen_time": 1700000012,
+        "version": "0.5.11",
+        "lora_freq": 906875,
+        "modem_preset": "LONG_FAST",
+    }
+    assert hb["version"] == "0.5.11"
+    assert hb["lora_freq"] == 906875
+
+
+def test_ingestor_heartbeat_without_optional_fields():
+    hb: IngestorHeartbeat = {
+        "node_id": "!aabbccdd",
+        "start_time": 1700000000,
+        "last_seen_time": 1700000013,
+        "version": "0.5.11",
+    }
+    assert "lora_freq" not in hb

tests/test_node_identity_unit.py

@@ -52,3 +52,23 @@ def test_node_num_from_id_parses_canonical_and_hex():
     assert node_num_from_id(123) == 123
 
 
+def test_canonical_node_id_rejects_none_and_empty():
+    assert canonical_node_id(None) is None
+    assert canonical_node_id("") is None
+    assert canonical_node_id("   ") is None
+
+
+def test_canonical_node_id_rejects_negative():
+    assert canonical_node_id(-1) is None
+    assert canonical_node_id(-0xABCDEF01) is None
+
+
+def test_canonical_node_id_truncates_overflow():
+    # Values wider than 32 bits are masked, not rejected.
+    assert canonical_node_id(0x1_ABCDEF01) == "!abcdef01"
+
+
+def test_node_num_from_id_rejects_none_and_empty():
+    assert node_num_from_id(None) is None
+    assert node_num_from_id("") is None
+    assert node_num_from_id("not-hex") is None

tests/test_provider_unit.py

@@ -26,11 +26,17 @@ if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
 from data.mesh_ingestor import daemon  # noqa: E402 - path setup
+from data.mesh_ingestor.provider import Provider  # noqa: E402 - path setup
 from data.mesh_ingestor.providers.meshtastic import (  # noqa: E402 - path setup
     MeshtasticProvider,
 )
 
 
+def test_meshtastic_provider_satisfies_protocol():
+    """MeshtasticProvider must structurally satisfy the Provider Protocol."""
+    assert isinstance(MeshtasticProvider(), Provider)
+
+
 def test_daemon_main_uses_provider_connect(monkeypatch):
     calls = {"connect": 0}
 
@@ -40,6 +46,7 @@ def test_daemon_main_uses_provider_connect(monkeypatch):
 
         def connect(self, *, active_candidate):  # type: ignore[override]
             calls["connect"] += 1
+
             # Return a minimal iface and stop immediately via Event.
             class Iface:
                 nodes = {}
@@ -89,13 +96,81 @@ def test_daemon_main_uses_provider_connect(monkeypatch):
         ),
     )
 
-    monkeypatch.setattr(daemon.handlers, "register_host_node_id", lambda *_a, **_k: None)
+    monkeypatch.setattr(
+        daemon.handlers, "register_host_node_id", lambda *_a, **_k: None
+    )
     monkeypatch.setattr(daemon.handlers, "host_node_id", lambda: "!host")
     monkeypatch.setattr(daemon.handlers, "upsert_node", lambda *_a, **_k: None)
     monkeypatch.setattr(daemon.handlers, "last_packet_monotonic", lambda: None)
-    monkeypatch.setattr(daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None)
-    monkeypatch.setattr(daemon.ingestors, "queue_ingestor_heartbeat", lambda *_a, **_k: True)
+    monkeypatch.setattr(
+        daemon.ingestors, "set_ingestor_node_id", lambda *_a, **_k: None
+    )
+    monkeypatch.setattr(
+        daemon.ingestors, "queue_ingestor_heartbeat", lambda *_a, **_k: True
+    )
 
     daemon.main(provider=FakeProvider())
     assert calls["connect"] >= 1
 
+
+def test_node_snapshot_items_retries_on_concurrent_mutation(monkeypatch):
+    """node_snapshot_items must retry on dict-mutation RuntimeError, not raise."""
+    from data.mesh_ingestor.providers.meshtastic import MeshtasticProvider
+
+    attempt = {"n": 0}
+
+    class MutatingNodes:
+        def items(self):
+            attempt["n"] += 1
+            if attempt["n"] < 3:
+                raise RuntimeError("dictionary changed size during iteration")
+            return [("!aabbccdd", {"num": 1})]
+
+    class FakeIface:
+        nodes = MutatingNodes()
+
+    monkeypatch.setattr("time.sleep", lambda _: None)
+    result = MeshtasticProvider().node_snapshot_items(FakeIface())
+    assert result == [("!aabbccdd", {"num": 1})]
+    assert attempt["n"] == 3
+
+
+def test_node_snapshot_items_returns_empty_after_retry_exhaustion(monkeypatch):
+    """node_snapshot_items returns [] (non-fatal) when all retries fail."""
+    from data.mesh_ingestor.providers.meshtastic import MeshtasticProvider
+    import data.mesh_ingestor.providers.meshtastic as _mod
+
+    class AlwaysMutating:
+        def items(self):
+            raise RuntimeError("dictionary changed size during iteration")
+
+    class FakeIface:
+        nodes = AlwaysMutating()
+
+    monkeypatch.setattr("time.sleep", lambda _: None)
+    monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None)
+    result = MeshtasticProvider().node_snapshot_items(FakeIface())
+    assert result == []
+
+
+def test_meshtastic_subscribe_is_idempotent(monkeypatch):
+    """Calling subscribe() twice returns the cached list without re-subscribing."""
+    import data.mesh_ingestor.providers.meshtastic as _m
+
+    subscribe_calls: list[str] = []
+
+    monkeypatch.setattr(
+        _m,
+        "pub",
+        types.SimpleNamespace(
+            subscribe=lambda _h, topic: subscribe_calls.append(topic)
+        ),
+    )
+
+    provider = MeshtasticProvider()
+    first = provider.subscribe()
+    second = provider.subscribe()
+
+    assert first == second
+    # pub.subscribe should only have been called once (first invocation)
+    assert len(subscribe_calls) == len(first)