fix: hold every background task, in both codebases

asyncio keeps only a weak reference to a task, so a coroutine started with
`asyncio.ensure_future(...)` whose result is discarded can be collected while
it is still running: the loop logs "Task was destroyed but it is pending!" and
the work simply does not happen. No error reaches the caller, and what is lost
is whatever that coroutine was in the middle of.

The node already had a guard for this, written after an abandoned stream task
lost a transcode slot for good — and it read one file, `webrtc_server.py`,
because that is where the defect was found. Outside that file there were
nineteen sites: the hub's `chat_notify` (a notification for every member of a
group), the indexer's debounce (every real-time index update), eleven in
`daemon.py` including the SIGHUP reload and each enrichment pass, two in
`ops.py`, and five in the loopback API.

`meshbay_common.background.spawn()` is the one door. It holds the task, drops
it when it finishes, and logs what it raised under the coroutine's own name —
an exception in a task nobody awaits was otherwise reported by asyncio at
collection time, out of context or not at all. A peer session's `_spawn` stays
as it is: that one can also *cancel* what it holds, which a module-level holder
cannot, because a session ends and a process does not.

`test_background_tasks.py` walks every package's source and refuses a discarded
handle. It parses rather than greps, so an assignment, a comprehension or an
await is not mistaken for one, and it was checked against a deliberate
reintroduction. A guard that stops at the edge of the file where the bug was
found is a guard against that bug, not against its class.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UMxEQadpzPkYLFf5CYKhpW

8 files changed, 258 insertions, 23 deletions

diff --git a/packages/meshbay-common/src/meshbay_common/background.py b/packages/meshbay-common/src/meshbay_common/background.py
new file mode 100644
index 0000000..1f25dfd
--- /dev/null
+++ b/packages/meshbay-common/src/meshbay_common/background.py
@@ -0,0 +1,70 @@
+"""
+Background work that outlives the call which started it.
+
+**asyncio keeps only a weak reference to a task.** A coroutine started with
+`asyncio.ensure_future(...)` whose result is thrown away can therefore be
+collected while it is still running: the loop logs "Task was destroyed but it is
+pending!" and the work simply does not happen. There is no error at the caller,
+nothing fails, and what was lost is whatever that coroutine was in the middle of
+— a notification never written, a file never indexed, a reload half applied.
+
+`spawn()` is the one door for that shape. The reference in `_running` is what
+keeps the task alive; the done callback is what stops the set growing.
+
+Two things it deliberately does **not** replace:
+
+- A task whose handle the caller keeps — the daemon's service tasks, the
+  indexer's reconcile loop — is already held, and being held is the whole point.
+- A task scoped to something that can end, where the owner must be able to
+  **cancel** it: a peer session's spawns die with the session
+  (`WebRTCPeerSession._spawn`), and an enrichment run is tracked by its own
+  module. Those keep their own sets, because a module-level one can only ever
+  grow and forget, never shut down.
+
+It also logs what a discarded task swallowed. An exception in a task nobody
+awaits is reported by asyncio at collection time, out of context and often not
+at all; here it is logged where it happened, naming the work.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Coroutine
+
+log = logging.getLogger(__name__)
+
+# Strong references to everything in flight. Module-level on purpose: these are
+# tasks with no owner to hold them, which is exactly why they need this.
+_running: set[asyncio.Task] = set()
+
+
+def spawn(coro: Coroutine[Any, Any, Any], *, what: str = "") -> asyncio.Task:
+    """Run a coroutine in the background and hold on to it until it finishes.
+
+    The work names itself in the log: `what` defaults to the coroutine's own
+    qualified name, so no call site has to carry a label for the sake of a line
+    nobody reads until something breaks. Pass one only where the coroutine's
+    name is not the useful answer.
+    """
+    if not what:
+        what = getattr(coro, "__qualname__", "") or repr(coro)
+    task = asyncio.ensure_future(coro)
+    _running.add(task)
+    task.add_done_callback(lambda t: _finished(t, what))
+    return task
+
+
+def _finished(task: asyncio.Task, what: str) -> None:
+    _running.discard(task)
+    if task.cancelled():
+        return
+    exc = task.exception()
+    if exc is not None:
+        log.error("Background task %s failed: %s", what or "(unnamed)", exc,
+                  exc_info=exc)
+
+
+def pending() -> int:
+    """How many background tasks are in flight. For diagnostics and tests."""
+    return len(_running)
diff --git a/packages/meshbay-common/tests/test_background_tasks.py b/packages/meshbay-common/tests/test_background_tasks.py
new file mode 100644
index 0000000..4f0ac59
--- /dev/null
+++ b/packages/meshbay-common/tests/test_background_tasks.py
@@ -0,0 +1,153 @@
+"""
+Tasks nobody holds, across every package.
+
+asyncio keeps only a **weak** reference to a task, so a coroutine started with
+`asyncio.ensure_future(...)` whose result is thrown away can be collected while
+it is still running. Nothing raises: the loop logs "Task was destroyed but it is
+pending!" and whatever that task was doing simply stops.
+
+The node already had a guard for this — written after an abandoned stream task
+never reached `async with sem.__aexit__` and lost a transcode slot for good —
+and it read one file, `webrtc_server.py`, because that is where the defect was
+found. Meanwhile the hub answered `chat_notify` with a discarded
+`ensure_future` (a notification for every member of a group, dropped whenever
+the collector got there first), the node's indexer scheduled every real-time
+index update the same way (a file copied in that never appears until a
+reconciliation sweep an unbounded time later), and seventeen more sites did it
+in `daemon.py`, `ops.py` and the loopback API — a reload, an enrichment pass, a
+swarm registration.
+
+So this one walks **every package's source**. A guard that stops at the edge of
+the file where the bug was found is a guard against that bug, not against its
+class.
+"""
+
+import ast
+import asyncio
+import logging
+from pathlib import Path
+
+import pytest
+
+from meshbay_common import background
+
+REPO = Path(__file__).resolve().parents[3]
+PACKAGES = ("meshbay-common", "meshbay-hub", "meshbay-node")
+
+SPAWNERS = {"ensure_future", "create_task"}
+
+# `spawn()` is the implementation of the rule and is the one place allowed to
+# call the primitive with nothing holding the result — that is what it is for.
+ALLOWED = {Path("meshbay-common/src/meshbay_common/background.py")}
+
+
+def _sources() -> list[Path]:
+    out: list[Path] = []
+    for pkg in PACKAGES:
+        src = REPO / "packages" / pkg / "src"
+        if src.is_dir():
+            out.extend(sorted(src.rglob("*.py")))
+    return out
+
+
+def test_there_are_sources_to_check():
+    """A rule applied to nothing passes forever."""
+    files = _sources()
+    assert len(files) > 20, f"only found {len(files)} source files — wrong root?"
+
+
+def _discarded(tree: ast.AST) -> list[int]:
+    """Lines where a task is started and its handle dropped on the floor.
+
+    An `Expr` statement is a call whose value goes nowhere; a `Lambda` whose
+    whole body is such a call is the same thing wearing a callback's clothes
+    (`signal.add_signal_handler(SIGHUP, lambda: ensure_future(reload()))`).
+    Anything assigned, returned, awaited or collected into a list is held by
+    its caller and is not this defect.
+    """
+    found: list[int] = []
+
+    def is_spawn(node: ast.AST) -> bool:
+        return (isinstance(node, ast.Call)
+                and isinstance(node.func, ast.Attribute)
+                and node.func.attr in SPAWNERS
+                and isinstance(node.func.value, ast.Name)
+                and node.func.value.id == "asyncio")
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Expr) and is_spawn(node.value):
+            found.append(node.lineno)
+        elif isinstance(node, ast.Lambda) and is_spawn(node.body):
+            found.append(node.lineno)
+    return found
+
+
+def test_no_task_is_started_and_forgotten():
+    offenders: list[str] = []
+    for path in _sources():
+        rel = path.relative_to(REPO / "packages")
+        if rel in ALLOWED:
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for line in _discarded(tree):
+            offenders.append(f"{rel}:{line}")
+
+    assert not offenders, (
+        "a task whose result is discarded can be collected mid-flight and the "
+        "work silently does not happen. Use meshbay_common.background.spawn(), "
+        "or keep the handle where the owner can also cancel it:\n  "
+        + "\n  ".join(offenders))
+
+
+# ── The helper itself ────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_spawn_holds_the_task_until_it_finishes():
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def work():
+        started.set()
+        await release.wait()
+
+    task = background.spawn(work())
+    await started.wait()
+    assert background.pending() == 1, "nothing is holding the task"
+    # The only strong reference the test itself holds, gone: if `spawn` is not
+    # keeping one, the task is now collectable.
+    del task
+    release.set()
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert background.pending() == 0, "the set grows for the life of the process"
+
+
+@pytest.mark.asyncio
+async def test_a_failure_is_logged_where_it_happened(caplog):
+    """Nobody awaits this task, so the log is the only place it can surface."""
+    async def work():
+        raise RuntimeError("the background work failed")
+
+    with caplog.at_level(logging.ERROR, logger=background.log.name):
+        # Pinned rather than assumed: an earlier test in the run may have
+        # raised this logger's level or disabled it outright.
+        background.log.disabled = False
+        background.log.setLevel(logging.NOTSET)
+        task = background.spawn(work(), what="the work")
+        with pytest.raises(RuntimeError):
+            await task
+
+    assert any("the work" in r.getMessage() for r in caplog.records), caplog.text
+
+
+@pytest.mark.asyncio
+async def test_the_work_names_itself_without_a_label():
+    async def a_named_coroutine():
+        raise RuntimeError("boom")
+
+    task = background.spawn(a_named_coroutine())
+    with pytest.raises(RuntimeError):
+        await task
+    # The default label is the coroutine's own name, so no call site has to
+    # carry one for a log line to be readable.
+    assert "a_named_coroutine" in a_named_coroutine.__qualname__
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 0e274f9..60b0c88 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -40,6 +40,7 @@ from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import jwt
+from meshbay_common.background import spawn
 from meshbay_hub.auth import hub_public_key_pem, decode_access_token
 from meshbay_hub.api.deps import get_current_user, require_admin
 from meshbay_hub.db.engine import get_db
@@ -397,7 +398,7 @@ async def node_websocket(ws: WebSocket):
                 if not _notify_budget(node_id):
                     log.warning("Node %s exceeded its chat_notify rate", node_id[:8])
                     continue
-                asyncio.ensure_future(_handle_chat_notify(
+                spawn(_handle_chat_notify(
                     msg.get("group_id", ""),
                     msg.get("sender_name", ""),
                     # The author, as the node authenticated them — not
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 9428ddd..518f221 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -37,6 +37,7 @@ from pathlib import Path
 
 import uvicorn
 
+from meshbay_common.background import spawn
 from meshbay_common.paths import fold
 from meshbay_common import MNP_VERSION
 from meshbay_common.protocol import MNP
@@ -738,7 +739,7 @@ class NodeDaemon:
             try:
                 loop.add_signal_handler(
                     signal.SIGHUP,
-                    lambda: asyncio.ensure_future(self._reload_config()))
+                    lambda: spawn(self._reload_config()))
             except (NotImplementedError, AttributeError):
                 pass          # no SIGHUP on Windows; `reload` says so there
             await stop_event.wait()
@@ -1257,7 +1258,7 @@ class NodeDaemon:
 
         def fire() -> None:
             self._pending_broadcasts.pop(group_id, None)
-            asyncio.ensure_future(self._broadcast_index_change(indexer))
+            spawn(self._broadcast_index_change(indexer))
 
         self._pending_broadcasts[group_id] = loop.call_later(
             self._broadcast_coalesce_secs, fire)
@@ -1309,14 +1310,14 @@ class NodeDaemon:
                 self._enriched_attempted.discard((group_id, entry.id))
             seen = {e.id for e in new_entries}
             new_entries = new_entries + [e for e in rebuilt if e.id not in seen]
-        asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries))
+        spawn(self._enrich_new_video_entries(indexer, new_entries))
         # Music app (docs/musicbay.md §6): same shape, gated on audio_root
         # exactly like video_root above (added later — musicbay.md's
         # original "no root, whole shared tree" call didn't hold up).
-        asyncio.ensure_future(self._enrich_new_audio_entries(indexer, new_entries))
+        spawn(self._enrich_new_audio_entries(indexer, new_entries))
         # Photos app (docs/photos.md §5): same shape, gated on photo_roots
         # (a list, not a single string — §2.1).
-        asyncio.ensure_future(self._enrich_new_photo_entries(indexer, new_entries))
+        spawn(self._enrich_new_photo_entries(indexer, new_entries))
 
         # A rename/move changes the very filename (or season folder) that
         # §3.3/§3.4's title-parse read display_title/season/episode from,
@@ -1327,11 +1328,11 @@ class NodeDaemon:
         # (duration/thumb_hash/... landing via _on_enriched below) leaves
         # name/path alone and must not re-trigger itself forever.
         if delta is not None and delta.updates and previous is not None:
-            asyncio.ensure_future(
+            spawn(
                 self._reenrich_renamed_video_entries(indexer, delta.updates, previous))
-            asyncio.ensure_future(
+            spawn(
                 self._reenrich_renamed_audio_entries(indexer, delta.updates, previous))
-            asyncio.ensure_future(
+            spawn(
                 self._reenrich_renamed_photo_entries(indexer, delta.updates, previous))
 
         # Videos/Music/Photos apps: a file that leaves the index also loses
@@ -1366,7 +1367,7 @@ class NodeDaemon:
                 still_referenced = any(
                     i.index.get_entry(file_id) is not None for i in self._indexers)
                 if not still_referenced:
-                    asyncio.ensure_future(self._media_cache.prune_file(file_id))
+                    spawn(self._media_cache.prune_file(file_id))
 
         # 11.5 — Push to connected WebRTC peers in this group
         if self._webrtc:
@@ -1404,7 +1405,7 @@ class NodeDaemon:
                      else [e.id for e in idx.entries])
             if hashes:
                 endpoint = f"webrtc:{self._config.node.quic_port}"
-                asyncio.ensure_future(self._register_swarm(hashes, endpoint))
+                spawn(self._register_swarm(hashes, endpoint))
 
     async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
         """
@@ -1666,7 +1667,7 @@ class NodeDaemon:
             return
         for session in list(self._webrtc._sessions.values()):
             if session._group_id == group_id:
-                asyncio.ensure_future(session.close())
+                spawn(session.close())
                 log.info("Dropped session for revoked group %s", group_id[:8])
 
     async def _register_swarm(self, hashes: list[str], endpoint: str) -> None:
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 852c061..da8ea82 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -35,6 +35,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from watchdog.events import FileSystemEvent, FileSystemEventHandler
 from watchdog.observers import Observer
 
+from meshbay_common.background import spawn
 from meshbay_common.paths import fold, find_fold_collisions, long_path
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.cache import IndexCache
@@ -933,7 +934,7 @@ class DirectoryIndexer:
 
         def fire() -> None:
             self._pending_timers.pop(key, None)
-            asyncio.ensure_future(self._update_entry(file_path, deleted))
+            spawn(self._update_entry(file_path, deleted))
 
         self._pending_timers[key] = self._loop.call_later(self.debounce_secs, fire)
 
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 4d3422d..91a922a 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -30,6 +30,7 @@ from dataclasses import asdict
 from pathlib import Path
 from typing import Any
 
+from meshbay_common.background import spawn
 from meshbay_common.chatbox import new_epoch_key
 from meshbay_common.crypto import (
     generate_gek,
@@ -1673,7 +1674,7 @@ async def set_app_directories(state: dict, group_id: str, app_key: str,
 
     enrich = (state.get("enrich_app_dirs_fns") or {}).get(app_key)
     if enrich:
-        asyncio.ensure_future(enrich(group_id))
+        spawn(enrich(group_id))
     return {"app": app_key, "directories": clean, "group_id": group_id}
 
 
@@ -1881,5 +1882,5 @@ async def start_reload(state: dict) -> dict:
     reload_fn = state.get("reload_fn")
     if not reload_fn:
         raise OpError("Reload not available", status=503)
-    asyncio.ensure_future(reload_fn())
+    spawn(reload_fn())
     return {"status": "reloading"}
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 4ad3787..de2542e 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -12,12 +12,12 @@ no server-rendered UI: the Node page ships in the desktop client (see
 `docs/refactor-node-ui.md`).
 """
 
-import asyncio
 import logging
 
 from fastapi import FastAPI, HTTPException, Query
 from fastapi.responses import JSONResponse
 
+from meshbay_common.background import spawn
 from meshbay_node import __version__, ops
 from meshbay_node.indexer.indexer import DirectoryIndexer
 
@@ -179,7 +179,7 @@ def create_ui_app(state: dict) -> FastAPI:
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
-            asyncio.ensure_future(reload_fn())
+            spawn(reload_fn())
         return result
 
     @app.post("/api/groups/detach")
@@ -190,7 +190,7 @@ def create_ui_app(state: dict) -> FastAPI:
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
-            asyncio.ensure_future(reload_fn())
+            spawn(reload_fn())
         return result
 
     @app.delete("/api/groups/{group_id}/files/{file_id}")
@@ -373,7 +373,7 @@ def create_ui_app(state: dict) -> FastAPI:
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
-            asyncio.ensure_future(reload_fn())
+            spawn(reload_fn())
         return result
 
     @app.patch("/api/groups/{group_id}/roots/{root_name}")
@@ -385,7 +385,7 @@ def create_ui_app(state: dict) -> FastAPI:
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
-            asyncio.ensure_future(reload_fn())
+            spawn(reload_fn())
         return result
 
     @app.put("/api/groups/{group_id}/roots/{root_name}/eject")
@@ -401,11 +401,11 @@ def create_ui_app(state: dict) -> FastAPI:
         result = await _op(lambda: ops.remove_root(state, group_id, root_name))
         reload_fn = state.get("reload_fn")
         if reload_fn:
-            asyncio.ensure_future(reload_fn())
+            spawn(reload_fn())
         return result
 
-    # asyncio.ensure_future above schedules the reload (and whatever initial
-    # scan it triggers) on the daemon's own event loop — it has no link to
+    # `spawn` above schedules the reload (and whatever initial scan it
+    # triggers) on the daemon's own event loop — it has no link to
     # this HTTP request or to any browser tab. Closing the client that made
     # this call does not cancel it: the scan is the node's own background
     # work, not something borrowed from the request that started it.
diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py
index 9dffedb..133c030 100644
--- a/packages/meshbay-node/tests/test_task_lifetime.py
+++ b/packages/meshbay-node/tests/test_task_lifetime.py
@@ -14,6 +14,14 @@ first try included, until the daemon was restarted. It was two slots at the
 time, which is how few it took.
 
 Seen in the wild on 2026-08-16 after a viewer switched films mid-stream.
+
+**What stays here is the session's own contract**: a peer session holds what it
+starts *and can cancel it* on the way out, which a module-level holder cannot —
+a session ends, a process does not. The same rule for every other
+fire-and-forget task, in every package, is
+`meshbay-common/tests/test_background_tasks.py`. This file used to state it for
+one file, and therefore did not state it for the nineteen sites outside that
+file.
 """
 
 import re