A supervised loop that crashes reports its node down

A push session that failed before it listened left only a task_crashed
event. The wrapper in _run_supervised now reports down with the exception,
unless the loop already said why it is down.

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

backend/fluksio/flow/nodes/base.py

@@ -90,6 +90,7 @@ class Node:
         "_pipeline",
         "synchronous",
         "_on_health",
+        "_health",
         "supervisor",
         "fingerprint",
     )
@@ -105,6 +106,9 @@ class Node:
         self.f = f
         self._pipeline: Pipeline | None = None
         self._on_health: Callable[[Node, str, str | None], None] | None = None
+        # What this node reported since its background loop last started, so
+        # a crash does not overwrite the reason the loop gave for being down.
+        self._health: str | 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
@@ -206,11 +210,25 @@ class Node:
 
         Returns the task only when unsupervised, since that is the one case
         where the caller has to cancel it itself.
+
+        A loop that raises reports the node down with the exception, unless it
+        already said why it is down. A push session that fails before it
+        listens would otherwise show up as a crash event and nowhere else.
         """
+
+        async def reported() -> None:
+            self._health = None
+            try:
+                await factory()
+            except Exception as exc:
+                if self._health != "down":
+                    self.report_health("down", f"{type(exc).__name__}: {exc}")
+                raise
+
         if self.supervisor is not None:
-            self.supervisor.spawn(f"{self.id}:{name}", self.flow, factory)
+            self.supervisor.spawn(f"{self.id}:{name}", self.flow, reported)
             return None
-        return asyncio.create_task(factory())
+        return asyncio.create_task(reported())
 
     @staticmethod
     async def _cancel_task(task: asyncio.Task[None]) -> None:
@@ -230,6 +248,7 @@ class Node:
 
     def report_health(self, status: str, detail: str | None = None) -> None:
         """Say how this node's connection is doing: ok, degraded or down."""
+        self._health = status
         if self._on_health is not None:
             self._on_health(self, status, detail)
 

backend/tests/flow/test_connector.py

@@ -14,6 +14,7 @@ from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import Node
 from fluksio.flow.pipeline import Pipeline
 from fluksio.flow.plugins import load_plugins
+from fluksio.flow.supervision import Supervisor
 
 
 class Sensor(ConnectorNode):
@@ -156,6 +157,35 @@ def test_an_undeclared_port_keeps_failing_until_the_node_declares_it():
     assert len([entry for entry in health if entry[0] == "down"]) > 1
 
 
+def test_a_push_session_that_fails_before_it_listens_reports_down():
+    """A crashed loop is node health, not only a crash event."""
+
+    class Push(Sensor):
+        async def start(self, app: Any = None) -> None:
+            self._run_supervised("push", self._session)
+
+        async def _session(self) -> None:
+            raise ConnectionError("bridge unreachable")
+
+    health: list[tuple[str, str | None]] = []
+    node = Push(provides=[MessageSpec(name="reading", dtype=DType.FLOAT)])
+    node.assign_flow("demo", "push")
+    node._on_health = lambda _node, status, detail: health.append((status, detail))
+
+    async def scenario() -> None:
+        node.supervisor = Supervisor()
+        await node.start()
+        for _ in range(500):
+            if health:
+                break
+            await asyncio.sleep(0.01)
+        await node.supervisor.cancel_all()
+
+    asyncio.run(scenario())
+
+    assert health == [("down", "ConnectionError: bridge unreachable")]
+
+
 class Actuator(ConnectorNode):
     """A connector that commands something instead of reading it."""