From 491678f75b5652f287ea4ad03c537b938f59b49d Mon Sep 17 00:00:00 2001
From: l5y <220195275+l5yth@users.noreply.github.com>
Date: Tue, 21 Apr 2026 11:41:44 +0200
Subject: [PATCH] data: catch packet handler errors (#759)

* data: catch packet handler errors

* data: address review comments

* data: address review comments
---
 data/mesh_ingestor/CONTRACTS.md               |   1 +
 .../protocols/_meshcore_patches.py            | 161 +++++++++
 data/mesh_ingestor/protocols/meshcore.py      |  61 ++++
 tests/test_meshcore_reader_patch.py           | 328 ++++++++++++++++++
 4 files changed, 551 insertions(+)
 create mode 100644 data/mesh_ingestor/protocols/_meshcore_patches.py
 create mode 100644 tests/test_meshcore_reader_patch.py

diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md
index 72133f7..4e7a2bb 100644
--- a/data/mesh_ingestor/CONTRACTS.md
+++ b/data/mesh_ingestor/CONTRACTS.md
@@ -80,6 +80,7 @@ The `v1:` prefix lets the format evolve (e.g. add a channel-secret hash) without
 - *Sender-side clock reset.* MeshCore nodes without an RTC start `sender_timestamp` from `0` after reboot. Two messages from the same sender containing the same text within one second of power-on collapse into a single row. Acceptable trade-off given the alternative (no dedup at all).
 - *Relay-rewritten `sender_timestamp` (#756).* MeshCore has been observed delivering the same physical packet twice with a rewritten `sender_timestamp` (≈10 s later, same `from_id`/`channel`/`text`), which flips the v1 fingerprint and bypasses the `messages.id` PK collapse. To cover this, the web app runs an additional content-level dedup on insert: for `protocol = "meshcore"` with non-empty `text` and a known `from_id`, a second row matching `(from_id, to_id, channel, text)` within ±30 s of `rx_time` is dropped (window lives in `MESHCORE_CONTENT_DEDUP_WINDOW_SECONDS`). The window is ~3× the observed relay delta; legitimate rapid re-sends of identical short text (e.g. `hi`, `ack`, `ok`, `test`) from the same sender on the same channel **within 30 s** will be silently collapsed into one row. Ingestors MUST still produce deterministic v1 ids — this content-level layer is additive, not a replacement. Pre-existing duplicates are cleared once by a `PRAGMA user_version`-gated one-shot backfill on startup.
 - *Concurrent-insert race (#756).* The content-dedup SELECT and the downstream INSERT are not currently wrapped in a shared transaction, so two concurrent Puma threads carrying the same content with different ids can both pass the pre-check and both insert. Duplicates produced this way are narrow (single-node multi-threaded ingest) and are not cleaned up on subsequent boots because the backfill is one-shot. If the race is ever observed in production, tighten `insert_message` to wrap the meshcore pre-check + id-PK path in `db.transaction(:immediate)`.
+- *Upstream `meshcore` reader crash on truncated advertisements (#754).* `meshcore-py` 2.3.6 (latest at the time of writing) raises `IndexError` from `MessageReader.handle_rx` at `reader.py:365` when a `DEVICE_INFO`/advertisement frame declares `fw_ver >= 10` but omits the trailing `path_hash_mode` byte. Because the frame is parsed inside a detached `asyncio.create_task(...)`, the exception surfaces as `Task exception was never retrieved` on stderr and the event for that frame is lost. The ingestor installs a runtime patch (`data/mesh_ingestor/protocols/_meshcore_patches.py`) that wraps `handle_rx`, logs one line with the first 32 bytes of the offending frame under `context=meshcore.reader.patch`, and lets the task exit cleanly; a loop-level handler (`context=asyncio.unhandled`) catches anything the targeted patch misses. Both shims are additive and will be removed once upstream ships a defensive length check.
 
 #### `POST /api/positions`
 
diff --git a/data/mesh_ingestor/protocols/_meshcore_patches.py b/data/mesh_ingestor/protocols/_meshcore_patches.py
new file mode 100644
index 0000000..f14b594
--- /dev/null
+++ b/data/mesh_ingestor/protocols/_meshcore_patches.py
@@ -0,0 +1,161 @@
+# 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.
+
+"""Runtime patches applied to the upstream ``meshcore`` library.
+
+This module exists solely to paper over bugs in the third-party
+``meshcore-py`` package while we wait for upstream fixes.  Each patch is
+narrow, idempotent, and preserves the original method on the target class so
+that it can be reverted cleanly once a fix ships upstream.
+
+Current patches:
+
+* :func:`_wrap_handle_rx` — guards :meth:`meshcore.reader.MessageReader.handle_rx`
+  against unhandled exceptions raised while decoding a single radio frame.
+  Upstream 2.3.6 (latest at the time of writing) raises ``IndexError`` at
+  ``reader.py:365`` when parsing a truncated ``DEVICE_INFO`` advertisement
+  (``path_hash_mode = dbuf.read(1)[0]`` with an already-exhausted buffer).
+  Because the frame is parsed inside a detached
+  ``asyncio.create_task(...)`` the resulting exception surfaces as a noisy
+  ``Task exception was never retrieved`` stderr dump and the decoded event
+  for that frame is lost.  See GitHub issue #754.
+
+Apply the patches by calling :func:`apply` as early as possible after the
+``meshcore`` package is imported.  Re-invoking :func:`apply` is a no-op.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .. import config
+
+# Sentinel attribute set on a patched method so repeated imports/tests do
+# not wrap the same function more than once.  The name intentionally
+# includes the project slug so we can grep for it while diagnosing.
+_PATCH_MARKER = "_potato_mesh_patched"
+
+# Cap on hex bytes dumped into the log per failure.  Keeps the log line
+# under a few hundred characters even for maximum-sized frames.
+_PACKET_LOG_MAX_BYTES = 32
+
+
+def apply() -> bool:
+    """Install every known-needed patch on the upstream ``meshcore`` library.
+
+    Safe to call multiple times; each patch is individually idempotent.
+
+    Implicit contract with upstream: every patch here rebinds a method on
+    the target *class*.  This only affects call sites that perform an
+    attribute lookup at call time (``reader.handle_rx(data)``) — not call
+    sites that captured an unbound reference before :func:`apply` ran
+    (``_rx = reader.handle_rx``).  As of ``meshcore-py`` 2.3.6 the library
+    always uses attribute-lookup-at-call, so this is fine; if a future
+    release flips that, the patch silently no-ops and the original bug
+    resurfaces.  Spot-check after every upstream bump.
+
+    Returns:
+        ``True`` when at least one patch was installed during this call,
+        ``False`` when every patch had already been applied (or when the
+        ``meshcore`` library is not importable in this environment, e.g. a
+        meshtastic-only test runner).
+    """
+    try:
+        import meshcore.reader as _reader  # type: ignore[import-not-found]
+    except ImportError:
+        # Meshtastic-only runtimes never load this module's caller, but
+        # imports from tests may still land here.  Nothing to patch.
+        return False
+
+    return _wrap_handle_rx(_reader.MessageReader)
+
+
+def _wrap_handle_rx(reader_cls: Any) -> bool:
+    """Wrap ``reader_cls.handle_rx`` with an exception-swallowing shim.
+
+    Parameters:
+        reader_cls: The ``MessageReader`` class to patch in place.
+
+    Returns:
+        ``True`` when the wrap was installed on this call; ``False`` when
+        the method had already been wrapped.
+    """
+    original = getattr(reader_cls, "handle_rx", None)
+    if original is None:
+        return False
+    if getattr(original, _PATCH_MARKER, False):
+        return False
+
+    async def safe_handle_rx(self, data, *args, **kwargs):  # type: ignore[no-untyped-def]
+        """Run the original ``handle_rx`` and convert hard failures to logs.
+
+        A single malformed frame would otherwise kill the
+        ``asyncio.create_task(reader.handle_rx(data))`` task spawned by the
+        upstream connection layer, surfacing as ``Task exception was never
+        retrieved`` in stderr and losing the event silently.  We log once
+        with the first few bytes of the offending frame for forensics and
+        then return ``None`` so the task exits cleanly.
+        """
+        try:
+            return await original(self, data, *args, **kwargs)
+        except Exception as exc:  # noqa: BLE001 — deliberately broad: a
+            # single malformed frame must not kill the reader.  Narrower
+            # excepts would hide future upstream failure modes (e.g.
+            # ``struct.error``) the same way the current IndexError was
+            # hidden before we added this shim.
+            config._debug_log(
+                "Suppressed meshcore reader exception on malformed frame",
+                context="meshcore.reader.patch",
+                severity="warning",
+                always=True,
+                error_class=type(exc).__name__,
+                error_message=str(exc),
+                packet_len=_safe_len(data),
+                packet_hex=_hex_preview(data, _PACKET_LOG_MAX_BYTES),
+            )
+            return None
+
+    setattr(safe_handle_rx, _PATCH_MARKER, True)
+    # Preserve the pre-patch method under a stable name so operators and
+    # future maintainers can revert the patch with one line.
+    reader_cls._orig_handle_rx = original
+    reader_cls.handle_rx = safe_handle_rx
+    return True
+
+
+def _safe_len(data: Any) -> int | None:
+    """Return ``len(data)`` or ``None`` when the object is not sized."""
+    try:
+        return len(data)
+    except TypeError:
+        return None
+
+
+def _hex_preview(data: Any, limit: int) -> str:
+    """Return the first *limit* bytes of ``data`` as a lowercase hex string.
+
+    Accepts anything that is a :class:`bytes`-like or supports ``bytes(data)``.
+    On conversion failure returns an empty string — the log caller still gets
+    the error class and message.
+    """
+    try:
+        if not isinstance(data, (bytes, bytearray, memoryview)):
+            data = bytes(data)
+    except Exception:  # noqa: BLE001 — pure diagnostic path, never raise.
+        return ""
+    prefix = bytes(data[:limit])
+    return prefix.hex()
+
+
+__all__ = ["apply"]
diff --git a/data/mesh_ingestor/protocols/meshcore.py b/data/mesh_ingestor/protocols/meshcore.py
index 1f3d46b..8963822 100644
--- a/data/mesh_ingestor/protocols/meshcore.py
+++ b/data/mesh_ingestor/protocols/meshcore.py
@@ -70,6 +70,21 @@ from meshcore import (
     TCPConnection,
 )
 
+from . import _meshcore_patches
+
+# Apply upstream-library patches before any ``MeshCore`` instance is built,
+# otherwise the first malformed advertisement dies inside a detached asyncio
+# task before our handler can observe it.  See
+# :mod:`data.mesh_ingestor.protocols._meshcore_patches` for the specific
+# upstream bugs covered.
+#
+# This mutates the upstream class at import time.  The blast radius is
+# narrow because ``protocols/__init__.py`` exposes this module only through
+# a lazy ``__getattr__`` and the daemon resolves it only when
+# ``PROTOCOL=meshcore`` is active.  Any future diagnostic CLI that imports
+# this module will inherit the shim.
+_meshcore_patches.apply()
+
 from .. import config, ingestors as _ingestors, queue as _queue
 from ..connection import default_serial_targets, parse_ble_target, parse_tcp_target
 from ..serialization import _iso, _node_num_from_id
@@ -1060,6 +1075,46 @@ def _make_connection(target: str, baudrate: int) -> object:
     return SerialConnection(target, baudrate)
 
 
+def _log_unhandled_loop_exception(
+    loop: asyncio.AbstractEventLoop, context: dict
+) -> None:
+    """Route asyncio's "unhandled task exception" warnings through our logger.
+
+    The upstream ``meshcore`` library spawns detached
+    ``asyncio.create_task`` tasks for every inbound radio frame.  When one
+    of those tasks raises and nobody awaits the future, asyncio's default
+    handler writes ``Task exception was never retrieved`` to stderr.  That
+    bypasses our structured log pipeline and clutters container logs.
+    This handler preserves the same information under
+    ``context=asyncio.unhandled`` so operators grep for one place.
+
+    Parameters:
+        loop: Event loop that surfaced the exception (unused but required
+            by the asyncio handler signature).
+        context: Asyncio exception-context dictionary.  Fields we care
+            about: ``message`` (human summary) and ``exception`` (the raw
+            exception object, when available).
+    """
+    del loop
+    exception = context.get("exception")
+    task = context.get("task")
+    task_name = None
+    if task is not None:
+        # Prefer the friendly ``get_name()``; fall back to ``repr`` for any
+        # future Task-like object that does not implement it.
+        get_name = getattr(task, "get_name", None)
+        task_name = get_name() if callable(get_name) else repr(task)
+    config._debug_log(
+        context.get("message") or "Unhandled asyncio task exception",
+        context="asyncio.unhandled",
+        severity="error",
+        always=True,
+        error_class=type(exception).__name__ if exception else None,
+        error_message=str(exception) if exception else None,
+        task=task_name,
+    )
+
+
 async def _run_meshcore(
     iface: _MeshcoreInterface,
     target: str,
@@ -1254,6 +1309,12 @@ class MeshcoreProvider:
         def _run_loop() -> None:
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)
+            # Second line of defence around issue #754: if a detached task
+            # inside the upstream ``meshcore`` library ever raises an
+            # exception we do not anticipate in ``_meshcore_patches``, funnel
+            # it through our logger instead of the default handler (which
+            # only writes ``Task exception was never retrieved`` to stderr).
+            loop.set_exception_handler(_log_unhandled_loop_exception)
             iface._loop = loop
             try:
                 loop.run_until_complete(
diff --git a/tests/test_meshcore_reader_patch.py b/tests/test_meshcore_reader_patch.py
new file mode 100644
index 0000000..075a49e
--- /dev/null
+++ b/tests/test_meshcore_reader_patch.py
@@ -0,0 +1,328 @@
+# 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 the runtime patch installed against the upstream ``meshcore``
+library to suppress ``MessageReader.handle_rx`` crashes on malformed frames.
+
+Covers GitHub issue #754.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+
+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.protocols import (  # noqa: E402 - path setup
+    _meshcore_patches,
+)
+
+
+class _FakeReader:
+    """Stand-in for ``meshcore.reader.MessageReader`` that lets us control
+    what ``handle_rx`` raises without dragging in the real library's
+    framing state machine."""
+
+    def __init__(self, raise_exc: BaseException | None = None, return_value=None):
+        self._raise_exc = raise_exc
+        self._return_value = return_value
+        self.received: list[bytes] = []
+
+    async def handle_rx(self, data):
+        self.received.append(bytes(data))
+        if self._raise_exc is not None:
+            raise self._raise_exc
+        return self._return_value
+
+
+def _install_patch_on(cls) -> None:
+    """Run the real wrap helper against an arbitrary class so the tests do
+    not mutate the installed ``meshcore.reader.MessageReader``."""
+    _meshcore_patches._wrap_handle_rx(cls)
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+def test_apply_is_idempotent():
+    """``apply()`` wrapping twice must not double-wrap the target method."""
+
+    class Target:
+        async def handle_rx(self, data):
+            return "ok"
+
+    first_wrap = _meshcore_patches._wrap_handle_rx(Target)
+    second_wrap = _meshcore_patches._wrap_handle_rx(Target)
+
+    assert first_wrap is True
+    assert second_wrap is False
+    # Marker is present on the wrapper so future imports short-circuit.
+    assert getattr(Target.handle_rx, _meshcore_patches._PATCH_MARKER, False) is True
+    # Original is preserved for revert.
+    assert hasattr(Target, "_orig_handle_rx")
+
+
+def test_apply_returns_false_when_already_patched(monkeypatch):
+    """Once ``_wrap_handle_rx`` has been applied, ``apply()`` at module
+    level observes the sentinel and short-circuits rather than rewrapping."""
+
+    class Target:
+        async def handle_rx(self, data):
+            return None
+
+    _meshcore_patches._wrap_handle_rx(Target)
+
+    # Replace ``meshcore.reader.MessageReader`` with our pre-patched Target
+    # so ``apply()`` cannot accidentally wrap a real class in the test env.
+    import meshcore.reader as reader_module
+
+    original_cls = reader_module.MessageReader
+    monkeypatch.setattr(reader_module, "MessageReader", Target)
+    try:
+        assert _meshcore_patches.apply() is False
+    finally:
+        monkeypatch.setattr(reader_module, "MessageReader", original_cls)
+
+
+def test_index_error_swallowed_and_logged(monkeypatch):
+    """The exact failure mode reported in #754: ``IndexError`` on a malformed
+    frame must not propagate and must emit one structured warning."""
+
+    class Target(_FakeReader):
+        pass
+
+    _install_patch_on(Target)
+    instance = Target(raise_exc=IndexError("index out of range"))
+
+    # Force the debug logger to always emit so we can capture the log line
+    # regardless of the ``DEBUG`` env flag during test runs.
+    from data.mesh_ingestor import config
+
+    emitted: list[tuple[str, dict]] = []
+
+    def _capture_log(message, **kwargs):
+        emitted.append((message, kwargs))
+
+    monkeypatch.setattr(config, "_debug_log", _capture_log)
+
+    # Should return None rather than raise.
+    result = _run(instance.handle_rx(b"\x01\x02\x03\x04"))
+    assert result is None
+
+    assert emitted, "patched handle_rx should have logged the suppressed error"
+    message, kwargs = emitted[-1]
+    assert "malformed frame" in message
+    assert kwargs["context"] == "meshcore.reader.patch"
+    assert kwargs["severity"] == "warning"
+    assert kwargs["error_class"] == "IndexError"
+    assert kwargs["error_message"] == "index out of range"
+    assert kwargs["packet_len"] == 4
+    assert kwargs["packet_hex"] == "01020304"
+
+
+def test_unrelated_return_value_preserved():
+    """When the original ``handle_rx`` returns normally, the wrapper must
+    forward the exact return value and not swallow it."""
+
+    class Target(_FakeReader):
+        pass
+
+    _install_patch_on(Target)
+    sentinel = object()
+    instance = Target(return_value=sentinel)
+
+    result = _run(instance.handle_rx(b"\x00"))
+    assert result is sentinel
+    assert instance.received == [b"\x00"]
+
+
+def test_packet_dump_truncated_to_max(monkeypatch):
+    """Large frames must be truncated in the hex dump so a noisy device
+    cannot flood the log."""
+
+    class Target(_FakeReader):
+        pass
+
+    _install_patch_on(Target)
+    instance = Target(raise_exc=ValueError("boom"))
+
+    from data.mesh_ingestor import config
+
+    emitted: list[dict] = []
+
+    def _capture_log(message, **kwargs):
+        emitted.append(kwargs)
+
+    monkeypatch.setattr(config, "_debug_log", _capture_log)
+
+    payload = bytes(range(256)) * 2  # 512 bytes
+    result = _run(instance.handle_rx(payload))
+    assert result is None
+
+    kwargs = emitted[-1]
+    # Hex length is exactly 2 * cap bytes.
+    expected_len = 2 * _meshcore_patches._PACKET_LOG_MAX_BYTES
+    assert len(kwargs["packet_hex"]) == expected_len
+    # And matches the first N real bytes of the payload.
+    assert (
+        kwargs["packet_hex"] == payload[: _meshcore_patches._PACKET_LOG_MAX_BYTES].hex()
+    )
+    assert kwargs["packet_len"] == 512
+
+
+def test_hex_preview_handles_non_bytes():
+    """Defensive: ``_hex_preview`` accepts bytearray / memoryview and any
+    object convertible via ``bytes(...)`` without raising."""
+
+    assert (
+        _meshcore_patches._hex_preview(bytearray(b"\xde\xad\xbe\xef"), 4) == "deadbeef"
+    )
+    assert _meshcore_patches._hex_preview(memoryview(b"\x01\x02"), 8) == "0102"
+    assert _meshcore_patches._hex_preview("not-bytes", 4) == ""
+
+
+def test_safe_len_handles_unsized():
+    assert _meshcore_patches._safe_len(b"\x01\x02") == 2
+    assert _meshcore_patches._safe_len(12345) is None
+
+
+def test_apply_skips_gracefully_when_meshcore_missing(monkeypatch):
+    """If ``meshcore`` is not importable, ``apply()`` must return ``False``
+    instead of raising.  Simulated by injecting an ImportError into
+    ``meshcore.reader``'s import machinery."""
+
+    # Block the import by clearing both the submodule and the parent, so
+    # that ``import meshcore.reader`` inside ``apply()`` triggers a fresh
+    # resolution that fails.
+    monkeypatch.setitem(sys.modules, "meshcore.reader", None)
+    assert _meshcore_patches.apply() is False
+
+
+def test_run_loop_exception_handler_routes_to_debug_log(monkeypatch):
+    """The loop-level safety net installed in ``_run_loop`` must forward
+    asyncio's unhandled-exception contexts through ``config._debug_log``."""
+
+    from data.mesh_ingestor import config
+    from data.mesh_ingestor.protocols import meshcore
+
+    emitted: list[tuple[str, dict]] = []
+
+    def _capture_log(message, **kwargs):
+        emitted.append((message, kwargs))
+
+    monkeypatch.setattr(config, "_debug_log", _capture_log)
+
+    loop = asyncio.new_event_loop()
+    try:
+        meshcore._log_unhandled_loop_exception(
+            loop,
+            {"message": "synthetic task failure", "exception": RuntimeError("boom")},
+        )
+    finally:
+        loop.close()
+
+    assert emitted, "loop handler should forward to the structured logger"
+    message, kwargs = emitted[-1]
+    assert message == "synthetic task failure"
+    assert kwargs["context"] == "asyncio.unhandled"
+    assert kwargs["severity"] == "error"
+    assert kwargs["error_class"] == "RuntimeError"
+    assert kwargs["error_message"] == "boom"
+
+
+def test_wrap_returns_false_when_class_has_no_handle_rx():
+    """If a future upstream release renames ``handle_rx`` or we point the
+    patch at the wrong class, ``_wrap_handle_rx`` must report the no-op
+    rather than silently install nothing on a random attribute."""
+
+    class Bare:
+        pass
+
+    assert _meshcore_patches._wrap_handle_rx(Bare) is False
+    assert not hasattr(Bare, "_orig_handle_rx")
+
+
+def test_loop_handler_defaults_when_context_minimal(monkeypatch):
+    """Covers the fallback branches of ``_log_unhandled_loop_exception`` —
+    missing ``message`` (defaults to a fixed string) and missing
+    ``exception`` (``error_class``/``error_message`` come through as ``None``).
+    Both are real asyncio code paths: task-cancellation and unhandled-future
+    warnings arrive with one-or-the-other key unset."""
+
+    from data.mesh_ingestor import config
+    from data.mesh_ingestor.protocols import meshcore
+
+    emitted: list[tuple[str, dict]] = []
+
+    def _capture_log(message, **kwargs):
+        emitted.append((message, kwargs))
+
+    monkeypatch.setattr(config, "_debug_log", _capture_log)
+
+    loop = asyncio.new_event_loop()
+    try:
+        # Empty context exercises both fallbacks at once.
+        meshcore._log_unhandled_loop_exception(loop, {})
+    finally:
+        loop.close()
+
+    assert emitted, "loop handler should still emit something for a bare context"
+    message, kwargs = emitted[-1]
+    assert message == "Unhandled asyncio task exception"
+    assert kwargs["context"] == "asyncio.unhandled"
+    assert kwargs["severity"] == "error"
+    assert kwargs["error_class"] is None
+    assert kwargs["error_message"] is None
+
+
+def test_loop_handler_logs_task_name_when_present(monkeypatch):
+    """Asyncio includes the failing ``task`` object in its context dict when
+    the exception comes from ``create_task(...)``.  The handler extracts the
+    task's name so operators can correlate log lines with the frame that
+    blew up when several readers share a loop."""
+
+    from data.mesh_ingestor import config
+    from data.mesh_ingestor.protocols import meshcore
+
+    emitted: list[dict] = []
+
+    def _capture_log(message, **kwargs):
+        emitted.append(kwargs)
+
+    monkeypatch.setattr(config, "_debug_log", _capture_log)
+
+    async def _dummy():
+        return None
+
+    loop = asyncio.new_event_loop()
+    try:
+        task = loop.create_task(_dummy(), name="meshcore-reader-42")
+        # Let the task finish so we don't leak a pending future.
+        loop.run_until_complete(task)
+        meshcore._log_unhandled_loop_exception(
+            loop,
+            {
+                "message": "synthetic",
+                "exception": ValueError("bad frame"),
+                "task": task,
+            },
+        )
+    finally:
+        loop.close()
+
+    assert emitted[-1]["task"] == "meshcore-reader-42"