Supervise the loops a flow starts, and give up loudly

A node's subscription, schedule or poll loop was a bare asyncio task: one
that raised outside its own retry handling was simply gone, and the node
went on being listed as running while nothing listened any more.

Those loops now run under a supervisor that restarts them with a growing
delay and quarantines a flow that burns through five restarts in five
minutes — a flow crash-looping every second is worse than one that is
visibly stopped, and the dashboard can now say which.

The MQTT subscription loses its private five-second reconnect in the
process: one backoff policy per socket, and it belongs to the supervisor.

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

ROADMAP.md

@@ -76,6 +76,10 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M
       reachability and fails the container healthcheck, so a wedged engine is restarted
       rather than counted as up. One engine per deployment — the API image runs a single
       worker, because a second one would be a second engine
+- [x] Supervised background tasks: a node's subscription, schedule or poll loop is
+      restarted with growing delay when it dies, and a flow that spends its failure
+      budget is quarantined and surfaced rather than left crash-looping. The loops
+      themselves no longer carry private retry logic
 - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking
       deployment on failure
 - [ ] User management scoped per flow and per data set

backend/app/api/routes/flows.py

@@ -172,6 +172,7 @@ def read_flows(controller: FlowControllerDep) -> Any:
                 has_draft=controller.store.has_draft(name),
                 enabled=controller.is_enabled(name),
                 paused=controller.is_paused(name),
+                quarantined=controller.is_quarantined(name),
             )
         )
     return FlowsPublic(data=summaries, count=len(summaries))

backend/app/flow/connector.py

@@ -109,21 +109,22 @@ class ConnectorNode(Node):
     # -------------------------------------------------------------------------
 
     async def start(self, app: FastAPI | None = None) -> None:
-        if self.config.poll_interval > 0 and self._poll_task is None:
+        if self.config.poll_interval > 0 and self._stop_event is None:
             self._stop_event = asyncio.Event()
-            self._poll_task = asyncio.create_task(self._poll_loop())
+            self._poll_task = self._run_supervised("poll", self._poll_loop)
 
     async def stop(self, app: FastAPI | None = None) -> None:
-        if self._poll_task is None:
+        if self._stop_event is None:
             return
-        if self._stop_event is not None:
-            self._stop_event.set()
-        self._poll_task.cancel()
-        try:
-            await self._poll_task
-        except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
-            pass
+        self._stop_event.set()
+        if self._poll_task is not None:
+            self._poll_task.cancel()
+            try:
+                await self._poll_task
+            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
+                pass
         self._poll_task = None
+        self._stop_event = None
         self._last_published = {}
 
     async def _poll_loop(self) -> None:

backend/app/flow/controller.py

@@ -43,6 +43,7 @@ from app.flow.schemas import (
 from app.flow.secrets import SecretNotFound, resolve_params
 from app.flow.state import MemoryState, StateBackend
 from app.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound
+from app.flow.supervision import Supervisor
 
 logger = logging.getLogger(__name__)
 
@@ -207,6 +208,7 @@ class FlowController:
         self.loaded: dict[str, LoadedNode] = {}
         self.issues: list[ValidationIssue] = []
         self.disabled: set[str] = set()
+        self.supervisor = Supervisor(events)
         self._lock = asyncio.Lock()
 
     # -------------------------------------------------------------------------
@@ -228,6 +230,9 @@ class FlowController:
         """Rebuild the whole pipeline from what is currently stored."""
         async with self._lock:
             await self._teardown()
+            # A fresh supervisor per build, so a flow quarantined by the last
+            # one gets another chance once its author has changed something.
+            self.supervisor = Supervisor(self.events)
 
             published = self.store.read_all()
             self.disabled = {
@@ -272,6 +277,8 @@ class FlowController:
                 await node.stop(self.app)
             except Exception:
                 logger.exception("Error stopping node '%s'", entry.id)
+        # After the nodes, so a loop still winding down is not restarted.
+        await self.supervisor.cancel_all()
 
     async def _activate(self) -> None:
         """Start subscriptions, schedules and webhooks of the new pipeline.
@@ -287,6 +294,7 @@ class FlowController:
             # that is what stopping it means.
             if entry.flow in self.disabled:
                 continue
+            node.supervisor = self.supervisor
             try:
                 await node.start(self.app)
             except Exception as exc:
@@ -474,6 +482,13 @@ class FlowController:
     def is_paused(self, flow: str) -> bool:
         return self.pipeline is not None and flow in self.pipeline.paused_flows()
 
+    def is_quarantined(self, flow: str) -> bool:
+        return flow in self.supervisor.quarantined
+
+    @property
+    def quarantined(self) -> set[str]:
+        return self.supervisor.quarantined
+
     def paused_flows(self) -> list[str]:
         return self.pipeline.paused_flows() if self.pipeline else []
 

backend/app/flow/nodes.py

@@ -5,7 +5,7 @@ import hmac
 import logging
 import threading
 import time
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Coroutine, Iterable
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Literal
 
@@ -15,6 +15,7 @@ from pydantic import BaseModel, ConfigDict, Field
 
 from app.flow import logs
 from app.flow.messages import MessageSpec, qualify
+from app.flow.supervision import Supervisor
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
@@ -75,6 +76,7 @@ class Node:
         "_pipeline",
         "synchronous",
         "_on_health",
+        "supervisor",
     )
 
     def __init__(
@@ -88,6 +90,9 @@ class Node:
         self.f = f
         self._pipeline: Pipeline | None = None
         self._on_health: Callable[[Node, str, str | None], None] | None = None
+        # Set by the controller before start(); a node built on its own (tests,
+        # previews) runs its loops unsupervised.
+        self.supervisor: Supervisor | None = None
         self.params = dict(params) if params else {}
         self.synchronous = bool(self.params.get("synchronous", False))
 
@@ -155,6 +160,19 @@ class Node:
     async def stop(self, app: FastAPI | None = None) -> None:
         """Undo :meth:`start`. Called before a rebuild, and must be idempotent."""
 
+    def _run_supervised(
+        self, name: str, factory: Callable[[], Coroutine[Any, Any, None]]
+    ) -> asyncio.Task | None:
+        """Run a background loop under supervision where there is one.
+
+        Returns the task only when unsupervised, since that is the one case
+        where the caller has to cancel it itself.
+        """
+        if self.supervisor is not None:
+            self.supervisor.spawn(f"{self.id}:{name}", self.flow, factory)
+            return None
+        return asyncio.create_task(factory())
+
     def report_health(self, status: str, detail: str | None = None) -> None:
         """Say how this node's connection is doing: ok, degraded or down."""
         if self._on_health is not None:
@@ -854,9 +872,7 @@ class MqttNode(Node):
         broker_host: str = "localhost"
         broker_port: int = 1883
         username: str | None = None
-        password: str | None = Field(
-            default=None, json_schema_extra={"x-secret": True}
-        )
+        password: str | None = Field(default=None, json_schema_extra={"x-secret": True})
         client_id: str | None = None
         qos: int = 0
         retain: bool = False
@@ -1071,7 +1087,7 @@ class MqttNode(Node):
             return  # Already running
 
         self._stop_event = asyncio.Event()
-        self._subscription_task = asyncio.create_task(self._subscription_loop())
+        self._subscription_task = self._run_supervised("mqtt", self._subscription_loop)
         logger.info(
             "Started MQTT subscription for node '%s' to topics %s",
             self.name,
@@ -1082,19 +1098,21 @@ class MqttNode(Node):
         """
         Stop the MQTT subscription.
 
-        Gracefully stops the background subscription task.
+        Gracefully stops the background subscription task. A supervised
+        subscription is cancelled with the rest of them at teardown; only an
+        unsupervised one is this method's to cancel.
         """
-        if self._subscription_task is None:
+        if self._stop_event is None:
             return
 
-        if self._stop_event:
-            self._stop_event.set()
+        self._stop_event.set()
 
-        self._subscription_task.cancel()
-        try:
-            await self._subscription_task
-        except asyncio.CancelledError:
-            pass
+        if self._subscription_task is not None:
+            self._subscription_task.cancel()
+            try:
+                await self._subscription_task
+            except asyncio.CancelledError:
+                pass
 
         self._subscription_task = None
         self._stop_event = None
@@ -1105,17 +1123,21 @@ class MqttNode(Node):
 
     async def _subscription_loop(self) -> None:
         """
-        Background loop that listens for MQTT messages and triggers the pipeline.
+        Listen for MQTT messages and trigger the pipeline.
 
         Subscribes to all unique topics from the ``topics`` mapping and
         uses the reverse lookup ``_topic_to_ports`` to route incoming
         payloads to the correct pipeline message names.
+
+        One connection attempt: a dropped broker raises, and the supervisor
+        decides when to try again. Reconnecting here as well would mean two
+        backoff policies fighting over the same socket.
         """
         import json
 
         import aiomqtt
 
-        while not (self._stop_event and self._stop_event.is_set()):
+        if not (self._stop_event and self._stop_event.is_set()):
             try:
                 async with aiomqtt.Client(
                     hostname=self.broker_host,
@@ -1189,27 +1211,25 @@ class MqttNode(Node):
                                 exc_info=True,
                             )
 
-            except aiomqtt.MqttError as e:
+                # Falling out of the message iterator without being told to
+                # stop means the broker went away quietly. Raising is how the
+                # supervisor hears about it.
+                if not (self._stop_event and self._stop_event.is_set()):
+                    self.report_health("down", "subscription ended")
+                    raise ConnectionError(
+                        f"MQTT subscription for '{self.name}' ended unexpectedly"
+                    )
+
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
                 logger.warning(
-                    "MQTT connection error in node '%s': %s. Retrying in 5s...",
-                    self.name,
-                    e,
+                    "MQTT subscription for node '%s' failed: %s", self.name, e
                 )
                 self.report_health("down", str(e))
-                if not (self._stop_event and self._stop_event.is_set()):
-                    # Reconnect after a delay
-                    await asyncio.sleep(5)
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.error(
-                    "Unexpected error in MQTT subscription for node '%s': %s. Retrying in 5s...",
-                    self.name,
-                    e,
-                    exc_info=True,
-                )
-                if not (self._stop_event and self._stop_event.is_set()):
-                    await asyncio.sleep(5)
+                if self._stop_event and self._stop_event.is_set():
+                    return
+                raise
 
     @property
     def is_subscribed(self) -> bool:
@@ -1880,22 +1900,22 @@ class DelayNode(Node):
             return
 
         self._stop_cron = asyncio.Event()
-        self._cron_task = asyncio.create_task(self._cron_loop())
+        self._cron_task = self._run_supervised("cron", self._cron_loop)
         logger.info("Started cron for node '%s': %s", self.name, self.cron_expr)
 
     async def stop_cron(self) -> None:
         """Stop the cron scheduler."""
-        if self._cron_task is None:
+        if self._stop_cron is None:
             return
 
-        if self._stop_cron:
-            self._stop_cron.set()
+        self._stop_cron.set()
 
-        self._cron_task.cancel()
-        try:
-            await self._cron_task
-        except asyncio.CancelledError:
-            pass
+        if self._cron_task is not None:
+            self._cron_task.cancel()
+            try:
+                await self._cron_task
+            except asyncio.CancelledError:
+                pass
 
         self._cron_task = None
         self._stop_cron = None
@@ -1955,16 +1975,14 @@ class DelayNode(Node):
                 await self._trigger_cron()
 
             except asyncio.CancelledError:
-                break
-            except Exception as e:
+                raise
+            except Exception:
+                # One bad tick should not cost the schedule; the supervisor
+                # picks it up from here, backoff included.
                 logger.error(
-                    "Error in cron loop for node '%s': %s",
-                    self.name,
-                    e,
-                    exc_info=True,
+                    "Error in cron loop for node '%s'", self.name, exc_info=True
                 )
-                # Back off on error to avoid tight loops
-                await asyncio.sleep(60)
+                raise
 
     async def _trigger_cron(self):
         """Emit data into the pipeline on a cron tick."""

backend/app/flow/schemas.py

@@ -126,6 +126,8 @@ class FlowSummary(BaseModel):
     has_draft: bool = False
     enabled: bool = True
     paused: bool = False
+    # Its background tasks kept crashing, so the engine stopped restarting them.
+    quarantined: bool = False
 
 
 class FlowsPublic(BaseModel):

backend/app/flow/supervision.py

@@ -0,0 +1,124 @@
+"""Supervision for the long-lived tasks a pipeline starts.
+
+A node's subscription, schedule or poll loop is an `asyncio.Task`, and a task
+that raises is simply gone — the node stays listed as running while nothing
+listens any more. The supervisor restarts those tasks with a growing delay and
+gives up on a flow that keeps failing, because a flow crash-looping every
+second is worse than a flow that is visibly stopped.
+
+Deliberately a plain task registry rather than a `TaskGroup`: a group cancels
+its siblings when one member fails, which is the opposite of what supervision
+means here.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections import deque
+from collections.abc import Callable, Coroutine
+from typing import Any
+
+from app.flow.events import EventBus
+
+logger = logging.getLogger(__name__)
+
+BACKOFF = (1.0, 5.0, 30.0, 60.0)
+# A flow that burns through this many restarts in the window is not going to
+# recover by being restarted again.
+FAILURE_BUDGET = 5
+FAILURE_WINDOW = 300.0
+
+TaskFactory = Callable[[], Coroutine[Any, Any, None]]
+
+
+class Supervisor:
+    """Keeps the pipeline's background tasks alive, or admits it cannot."""
+
+    def __init__(self, events: EventBus | None = None) -> None:
+        self._events = events
+        self._tasks: dict[str, asyncio.Task[None]] = {}
+        self._failures: dict[str, deque[float]] = {}
+        self.quarantined: set[str] = set()
+
+    def spawn(self, name: str, flow: str, factory: TaskFactory) -> None:
+        """Run `factory()` and keep running it until told to stop."""
+        if name in self._tasks:
+            return
+        self._tasks[name] = asyncio.create_task(
+            self._supervise(name, flow, factory), name=f"supervised:{name}"
+        )
+
+    async def _supervise(self, name: str, flow: str, factory: TaskFactory) -> None:
+        attempt = 0
+        while True:
+            try:
+                await factory()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                if not self._record_failure(name, flow, exc):
+                    return
+            else:
+                # A clean return means the loop decided it was done.
+                return
+
+            delay = BACKOFF[min(attempt, len(BACKOFF) - 1)]
+            attempt += 1
+            await asyncio.sleep(delay)
+
+    def _record_failure(self, name: str, flow: str, exc: Exception) -> bool:
+        """Note the crash; False when the flow has spent its budget."""
+        logger.warning("Supervised task '%s' crashed: %s", name, exc, exc_info=True)
+        self._publish(
+            {
+                "type": "task_crashed",
+                "task": name,
+                "flow": flow,
+                "error": f"{type(exc).__name__}: {exc}",
+                "ts": time.time(),
+            }
+        )
+
+        now = time.monotonic()
+        window = self._failures.setdefault(flow, deque())
+        window.append(now)
+        while window and window[0] < now - FAILURE_WINDOW:
+            window.popleft()
+
+        if len(window) < FAILURE_BUDGET:
+            return True
+
+        logger.error(
+            "Flow '%s' crashed %d times in %.0fs — quarantined",
+            flow,
+            len(window),
+            FAILURE_WINDOW,
+        )
+        self.quarantined.add(flow)
+        self._publish(
+            {
+                "type": "flow_quarantined",
+                "flow": flow,
+                "error": f"{type(exc).__name__}: {exc}",
+                "ts": time.time(),
+            }
+        )
+        return False
+
+    async def cancel_all(self) -> None:
+        """Stop supervising. Idempotent, and safe to call mid-restart."""
+        tasks = list(self._tasks.values())
+        self._tasks.clear()
+        for task in tasks:
+            task.cancel()
+        for task in tasks:
+            try:
+                await task
+            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
+                pass
+
+    def _publish(self, event: dict[str, Any]) -> None:
+        if self._events is not None:
+            self._events.publish(event)

backend/tests/flow/test_supervision.py

@@ -0,0 +1,126 @@
+"""Supervision: a background task that dies gets restarted, until it is hopeless."""
+
+import asyncio
+
+import pytest
+
+from app.flow import supervision
+from app.flow.events import EventBus
+from app.flow.supervision import FAILURE_BUDGET, Supervisor
+
+
+@pytest.fixture(autouse=True)
+def no_backoff(monkeypatch: pytest.MonkeyPatch):
+    """Real backoff would make these tests a minute long."""
+    monkeypatch.setattr(supervision, "BACKOFF", (0.0,))
+
+
+async def _settle() -> None:
+    """Let the supervisor work through its restarts."""
+    for _ in range(50):
+        await asyncio.sleep(0)
+
+
+def test_a_crashing_loop_is_restarted():
+    attempts = 0
+
+    async def loop() -> None:
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise ConnectionError("broker went away")
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("mqtt", "heating", loop)
+        await _settle()
+
+        assert attempts == 3
+        assert supervisor.quarantined == set()
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_a_loop_that_returns_is_left_alone():
+    """Returning is how a loop says it is finished, not that it failed."""
+    attempts = 0
+
+    async def loop() -> None:
+        nonlocal attempts
+        attempts += 1
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("cron", "heating", loop)
+        await _settle()
+
+        assert attempts == 1
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_a_flow_that_keeps_crashing_is_quarantined():
+    attempts = 0
+    events: list[dict] = []
+    bus = EventBus()
+    bus.publish = events.append  # type: ignore[method-assign]
+
+    async def loop() -> None:
+        nonlocal attempts
+        attempts += 1
+        raise RuntimeError("nope")
+
+    async def scenario() -> None:
+        supervisor = Supervisor(bus)
+        supervisor.spawn("mqtt", "heating", loop)
+        await _settle()
+
+        # Restarting stops once the budget is gone, rather than spinning forever.
+        assert attempts == FAILURE_BUDGET
+        assert supervisor.quarantined == {"heating"}
+        assert [e["type"] for e in events].count("task_crashed") == FAILURE_BUDGET
+        assert [e["type"] for e in events][-1] == "flow_quarantined"
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_the_budget_is_per_flow():
+    ran = 0
+
+    async def failing() -> None:
+        raise RuntimeError("nope")
+
+    async def healthy() -> None:
+        nonlocal ran
+        ran += 1
+        await asyncio.sleep(3600)
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("a", "broken", failing)
+        supervisor.spawn("b", "fine", healthy)
+        await _settle()
+
+        assert supervisor.quarantined == {"broken"}
+        assert ran == 1
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+
+def test_cancel_all_is_idempotent():
+    async def loop() -> None:
+        await asyncio.sleep(3600)
+
+    async def scenario() -> None:
+        supervisor = Supervisor()
+        supervisor.spawn("mqtt", "heating", loop)
+        await _settle()
+
+        await supervisor.cancel_all()
+        await supervisor.cancel_all()
+
+    asyncio.run(scenario())

frontend/src/client/schemas.gen.ts

@@ -364,6 +364,11 @@ export const FlowSummarySchema = {
             type: 'boolean',
             title: 'Paused',
             default: false
+        },
+        quarantined: {
+            type: 'boolean',
+            title: 'Quarantined',
+            default: false
         }
     },
     type: 'object',

frontend/src/client/types.gen.ts

@@ -101,6 +101,7 @@ export type FlowSummary = {
     has_draft?: boolean;
     enabled?: boolean;
     paused?: boolean;
+    quarantined?: boolean;
 };
 
 /**