Backend: real alert test results, state cleanup on delete/rename, node trigger errors, queue and collector fixes

`AlertManager.send` swallowed every delivery failure, so the alerts screen's
Test button answered 200 whatever happened — the one thing it exists for. It
takes `raise_on_error` now, which only the test route passes; the per-channel
loop keeps the swallow, because one dead channel must not stop the others
hearing about the same fault. A refused delivery answers 502 with whatever the
sender said.

Renaming a flow left its values under the old name for good: the delete path
already swept them, the rename path never did. It calls the same `forget_flow`,
which covers the messages and the `__ts__`/`__version__`/`__history__`
bookkeeping keyed by message name. Cleanup, not migration — they repopulate
under the new name on the next run.

Triggering a node by hand ran `Node.__call__` with nothing catching it, so a
node that raised produced a 500 and a stack trace in the server log, and
nothing at all on the canvas. `Pipeline.publish_error` is the reporting half of
`_execute_node` lifted out; both paths go through it, so a manual failure now
reads the same on the canvas and in the metrics as a queued one. The route
answers 400 with the node's error.

`MemoryWorkQueue.stats()` counts claimed-but-unacknowledged work rather than
reporting zero, so the health tile means something without Redis. The metrics
collector's held tracebacks are capped at `DETAIL_CAP` and swept on the same
`RUN_STALE_S` cutoff the open runs use, instead of one untruncated traceback
per node kept for the life of the process — a traceback still survives the
flush between the log and the failure it belongs to.

`GET /observability/events` takes `since`/`until`, the window `/runs` already
took, so a failures list can cover the span the charts beside it are drawn from.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XC2jX6Hdj7pxGGKzBTrbqB

backend/app/api/routes/alerts.py

@@ -74,12 +74,18 @@ async def test_channel(channel_name: str, controller: FlowControllerDep) -> Any:
     if channel is None:
         raise HTTPException(status_code=404, detail=f"No channel '{channel_name}'")
 
-    await controller.alerts.send(
+    try:
-        channel,
+        await controller.alerts.send(
-        Alert(
+            channel,
-            title="Fluksio test alert",
+            Alert(
-            body="If you are reading this, the channel works.",
+                title="Fluksio test alert",
-            severity="warning",
+                body="If you are reading this, the channel works.",
-        ),
+                severity="warning",
-    )
+            ),
+            raise_on_error=True,
+        )
+    except Exception as exc:
+        # Whatever the sender said, verbatim: it is the only clue the operator
+        # has about why the channel does not work.
+        raise HTTPException(status_code=502, detail=str(exc) or type(exc).__name__)
     return Message(message=f"Sent a test alert through '{channel_name}'")

backend/app/api/routes/flows.py

@@ -432,6 +432,10 @@ async def rename_flow(
     except FlowExists as exc:
         raise HTTPException(status_code=409, detail=str(exc))
 
+    # The old name is nobody's namespace now; its values would sit there under a
+    # flow that no longer exists. They repopulate under the new name on the next
+    # run, so this is cleanup rather than a migration.
+    await run_in_threadpool(controller.forget_flow, name)
     await controller.reload()
     return _detail(controller, renamed)
 
@@ -633,13 +637,17 @@ async def trigger_node(
     """Feed values into a single node."""
     _require_enabled(controller, name)
     try:
-        await run_in_threadpool(
+        error = await run_in_threadpool(
             controller.trigger_node, f"{name}.{node_id}", body.values
         )
     except KeyError:
         raise HTTPException(
             status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
         )
+    if error:
+        # The canvas already has the failure from the bus; the person who
+        # clicked gets to read it too, instead of a stack trace in the log.
+        raise HTTPException(status_code=400, detail=error)
     return _flow_state(controller, name)
 
 

backend/app/api/routes/observability.py

@@ -306,9 +306,15 @@ def read_events(
     session: SessionDep,
     kind: Literal["failure", "audit"] = "failure",
     flow: str | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
     limit: int = 100,
 ) -> Any:
-    """What went wrong, or who changed what. Newest first."""
+    """What went wrong, or who changed what. Newest first.
+
+    ``since`` is inclusive and ``until`` exclusive, the same window ``/runs``
+    takes, so a list can cover the span the charts beside it are drawn from.
+    """
     statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc())
     if kind == "audit":
         statement = statement.where(col(EngineEvent.type) == "audit")
@@ -316,6 +322,10 @@ def read_events(
         statement = statement.where(col(EngineEvent.type) != "audit")
     if flow:
         statement = statement.where(col(EngineEvent.flow) == flow)
+    if since:
+        statement = statement.where(col(EngineEvent.ts) >= _aware(since))
+    if until:
+        statement = statement.where(col(EngineEvent.ts) < _aware(until))
     return list(session.exec(statement.limit(min(limit, 500))))
 
 

backend/app/flow/alerts.py

@@ -271,10 +271,21 @@ class AlertManager:
     # Delivery
     # -------------------------------------------------------------------------
 
-    async def send(self, channel: Channel, alert: Alert) -> None:
+    async def send(
+        self, channel: Channel, alert: Alert, raise_on_error: bool = False
+    ) -> None:
+        """Deliver one alert.
+
+        A failure is logged and swallowed, because one dead channel must not
+        stop the others hearing about the same fault. The test button passes
+        ``raise_on_error``: telling a working channel from a broken one is the
+        only thing it exists for.
+        """
         try:
             config = resolve_params(channel.config)
         except Exception as exc:
+            if raise_on_error:
+                raise
             logger.error("Channel '%s' has unusable settings: %s", channel.name, exc)
             return
 
@@ -286,6 +297,8 @@ class AlertManager:
             else:
                 await self._send_email(config, alert)
         except Exception as exc:
+            if raise_on_error:
+                raise
             logger.error("Could not alert through '%s': %s", channel.name, exc)
 
     async def _send_ntfy(self, config: dict[str, Any], alert: Alert) -> None:

backend/app/flow/controller.py

@@ -836,19 +836,29 @@ class FlowController:
         )
         pipeline.run(inputs or {})
 
-    def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> None:
+    def trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> str:
         """Feed values into one node. Blocking — call from a worker thread.
 
         Runs here rather than through the queue: the caller is a person waiting
         on the response, and wants the state it produced.
+
+        Returns what the node raised, or an empty string. A node failing by hand
+        is reported the way a queued run reports it — on the canvas and in the
+        metrics — rather than as a stack trace in the server log.
         """
         node = self.get_node(node_id)
         if node is None:
             raise KeyError(node_id)
-        if node.requires and values:
+        try:
-            node.trigger(values, durable=False)
+            if node.requires and values:
-        else:
+                node.trigger(values, durable=False)
-            node.inject(values or {}, durable=False)
+            else:
+                node.inject(values or {}, durable=False)
+        except Exception as exc:
+            if self.pipeline is None:
+                raise
+            return self.pipeline.publish_error(node, exc)
+        return ""
 
     def _publish(self, event: dict[str, Any]) -> None:
         if self.events is not None:

backend/app/flow/metrics.py

@@ -87,8 +87,9 @@ class MetricsCollector:
         self._buckets: dict[tuple[str, str, datetime], dict[str, float]] = {}
         self._runs: dict[str, dict[str, Any]] = {}
         self._pending: list[EngineEvent] = []
-        # The traceback arrives one event before the failure it belongs to.
+        # The traceback arrives one event before the failure it belongs to,
-        self._tracebacks: dict[tuple[str, str], str] = {}
+        # held with the time it arrived so an unpaired one does not stay put.
+        self._tracebacks: dict[tuple[str, str], tuple[float, str]] = {}
         self._last_prune = 0.0
 
     # -------------------------------------------------------------------------
@@ -171,7 +172,7 @@ class MetricsCollector:
             # Held for the node_error that follows it from the same thread.
             if event.get("level") == "error":
                 key = (str(event.get("flow") or ""), str(event.get("node") or ""))
-                self._tracebacks[key] = str(event.get("text") or "")
+                self._tracebacks[key] = (ts, str(event.get("text") or "")[:DETAIL_CAP])
             return
 
         if kind == "node_error":
@@ -179,7 +180,7 @@ class MetricsCollector:
             if run is not None:
                 run["errors"] += 1
             key = (str(event.get("flow") or ""), str(event.get("node") or ""))
-            traceback = self._tracebacks.pop(key, "")
+            traceback = self._tracebacks.pop(key, (0.0, ""))[1]
             error = str(event.get("error") or "")
             self._pending.append(
                 EngineEvent(
@@ -293,7 +294,11 @@ class MetricsCollector:
                 del self._runs[run_id]
         # Held tracebacks survive the flush: the log and the failure it belongs
         # to are two events, and a flush can fall between them. One per node,
-        # each replaced by that node's next failure.
+        # each replaced by that node's next failure — and dropped once it is
+        # clear the failure it was waiting for is never coming.
+        for key, (held, _) in list(self._tracebacks.items()):
+            if held < cutoff:
+                del self._tracebacks[key]
 
     def _hold(
         self,

backend/app/flow/pipeline.py

@@ -569,6 +569,37 @@ class Pipeline:
             }
         )
 
+    def publish_error(
+        self,
+        node: Node,
+        exc: Exception,
+        collected: logs.Collector | None = None,
+        entry_id: str = "",
+    ) -> str:
+        """Report a node failure, and hand the one-line version back.
+
+        Call it from an ``except`` block — the traceback comes from the
+        exception being handled. Every way of running a node reports through
+        here, so a manual run reads the same on the canvas and in the metrics
+        as a queued one.
+        """
+        logger.exception("Node '%s' failed", node.id)
+        # The one-line error goes on the node; the traceback goes to the log
+        # panel, which is where there is room to read it.
+        self.publish_log(node, collected or logs.Collector(), logs.node_traceback())
+        error = f"{type(exc).__name__}: {exc}"
+        self._publish(
+            {
+                "type": "node_error",
+                "flow": node.flow,
+                "node": node.id,
+                "error": error,
+                "run": entry_id,
+                "ts": time.time(),
+            }
+        )
+        return error
+
     def _already_done(self, entry_id: str, node: Node) -> bool:
         """Did this node's side effect already happen for this work item?"""
         if node.idempotent or self._queue is None:
@@ -650,20 +681,7 @@ class Pipeline:
             return result
         except Exception as exc:
             # One failing node must not take the rest of the graph down.
-            logger.exception("Node '%s' failed", node.id)
+            self.publish_error(node, exc, collected, entry_id)
-            # The one-line error goes on the node; the traceback goes to the
-            # log panel, which is where there is room to read it.
-            self.publish_log(node, collected, logs.node_traceback())
-            self._publish(
-                {
-                    "type": "node_error",
-                    "flow": node.flow,
-                    "node": node.id,
-                    "error": f"{type(exc).__name__}: {exc}",
-                    "run": entry_id,
-                    "ts": time.time(),
-                }
-            )
             return None
 
     # -------------------------------------------------------------------------

backend/app/flow/queue.py

@@ -177,6 +177,8 @@ class MemoryWorkQueue(WorkQueue):
         self._delayed: list[tuple[float, int, WorkItem]] = []
         self._parked: dict[str, list[WorkItem]] = {}
         self._done: set[tuple[str, str]] = set()
+        # Claimed and not yet acknowledged, which is what Redis's `pending` is.
+        self._in_flight = 0
         self._counter = 0
         self._seq = 0
         self._lock = threading.Lock()
@@ -210,10 +212,16 @@ class MemoryWorkQueue(WorkQueue):
                 if remaining <= 0:
                     return []
                 self._wake.wait(remaining)
-            return [self._items.popleft() for _ in range(min(count, len(self._items)))]
+            claimed = [
+                self._items.popleft() for _ in range(min(count, len(self._items)))
+            ]
+            self._in_flight += len(claimed)
+            return claimed
 
     def ack(self, item: WorkItem) -> None:
-        """Nothing to acknowledge: claiming already removed it."""
+        """Nothing to take off the queue — claiming did that. Only counted."""
+        with self._lock:
+            self._in_flight = max(0, self._in_flight - 1)
 
     def reclaim_stale(self, min_idle_ms: int) -> list[WorkItem]:
         """Nothing to reclaim: an item lost here died with the process."""
@@ -253,7 +261,7 @@ class MemoryWorkQueue(WorkQueue):
         with self._lock:
             return {
                 "depth": len(self._items),
-                "pending": 0,
+                "pending": self._in_flight,
                 "delayed": len(self._delayed),
                 "parked": sum(len(v) for v in self._parked.values()),
                 "oldest_pending_s": 0.0,

backend/tests/api/routes/test_alerts.py

@@ -0,0 +1,66 @@
+"""The Test button: it exists to tell a working channel from a broken one."""
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+from app.flow.alerts import AlertsConfig, Channel
+
+PREFIX = f"{settings.API_V1_STR}/alerts"
+
+
+def _one_channel(client: TestClient) -> Any:
+    alerts = client.app.state.flow_controller.alerts  # type: ignore[attr-defined]
+    alerts.config = AlertsConfig(
+        channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})]
+    )
+    return alerts
+
+
+def test_a_channel_that_cannot_deliver_is_not_a_success(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    alerts = _one_channel(client)
+
+    async def refuse(_config: dict[str, Any], _alert: Any) -> None:
+        raise RuntimeError("nobody home")
+
+    monkeypatch.setattr(alerts, "_send_ntfy", refuse)
+
+    response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers)
+
+    assert response.status_code == 502
+    assert "nobody home" in response.json()["detail"]
+
+
+def test_a_channel_that_delivers_says_so(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    alerts = _one_channel(client)
+    sent = []
+
+    async def accept(_config: dict[str, Any], alert: Any) -> None:
+        sent.append(alert)
+
+    monkeypatch.setattr(alerts, "_send_ntfy", accept)
+
+    response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers)
+
+    assert response.status_code == 200
+    assert len(sent) == 1
+
+
+def test_an_unknown_channel_is_a_404(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    _one_channel(client)
+
+    response = client.post(f"{PREFIX}/test/nowhere", headers=superuser_token_headers)
+
+    assert response.status_code == 404

backend/tests/api/routes/test_flows.py

@@ -357,3 +357,69 @@ def test_rename_rejects_an_invalid_name(
         json={"new_name": "Not A Flow Name"},
     )
     assert response.status_code == 400
+
+
+def test_a_node_that_raises_answers_with_its_error(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """A manual run fails the way a scheduled one does, not as a 500."""
+    saved = client.put(
+        f"{PREFIX}/failing", headers=superuser_token_headers, json=a_flow("failing")
+    ).json()
+    client.put(
+        f"{PREFIX}/failing/nodes/sensor/source",
+        headers=superuser_token_headers,
+        json={"code": BROKEN_NODE},
+    )
+    client.post(
+        f"{PREFIX}/failing/publish",
+        headers=superuser_token_headers,
+        json={"version": saved["definition"]["version"]},
+    )
+
+    response = client.post(
+        f"{PREFIX}/failing/nodes/sensor/trigger",
+        headers=superuser_token_headers,
+        json={"values": {}},
+    )
+
+    assert response.status_code == 400
+    assert "boom" in response.json()["detail"]
+
+    client.delete(f"{PREFIX}/failing", headers=superuser_token_headers)
+
+
+def test_renaming_a_flow_leaves_nothing_under_the_old_name(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    saved = client.put(
+        f"{PREFIX}/movable", headers=superuser_token_headers, json=a_flow("movable")
+    ).json()
+    client.put(
+        f"{PREFIX}/movable/nodes/sensor/source",
+        headers=superuser_token_headers,
+        json={"code": WORKING_NODE},
+    )
+    client.post(
+        f"{PREFIX}/movable/publish",
+        headers=superuser_token_headers,
+        json={"version": saved["definition"]["version"]},
+    )
+    client.post(
+        f"{PREFIX}/movable/run", headers=superuser_token_headers, json={"inputs": {}}
+    )
+
+    state = client.app.state.flow_controller.state
+    assert [key for key in state.keys() if "movable." in key]
+
+    client.post(
+        f"{PREFIX}/movable/rename",
+        headers=superuser_token_headers,
+        json={"new_name": "moved"},
+    )
+
+    # The values repopulate under the new name on the next run; what the old
+    # name left behind would sit there for good.
+    assert [key for key in state.keys() if "movable." in key] == []
+
+    client.delete(f"{PREFIX}/moved", headers=superuser_token_headers)

backend/tests/api/routes/test_observability.py

@@ -134,3 +134,35 @@ def test_runs_narrow_to_one_minute(
     # The upper bound is exclusive, so the run starting the next minute is not
     # in this one.
     assert [run["id"] for run in runs] == ["minute-in"]
+
+
+def test_events_narrow_to_one_minute(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """The failures list reaches as far back as the charts beside it."""
+    minute = datetime.now(timezone.utc).replace(second=0, microsecond=0) - timedelta(
+        hours=4
+    )
+    db.add(EngineEvent(ts=minute, type="node_error", flow=FLOW, detail="minute-in"))
+    db.add(
+        EngineEvent(
+            ts=minute + timedelta(minutes=1),
+            type="node_error",
+            flow=FLOW,
+            detail="minute-after",
+        )
+    )
+    db.commit()
+
+    events = client.get(
+        f"{PREFIX}/events",
+        headers=superuser_token_headers,
+        params={
+            "flow": FLOW,
+            "since": minute.isoformat(),
+            "until": (minute + timedelta(minutes=1)).isoformat(),
+        },
+    ).json()
+
+    # The upper bound is exclusive, so the failure a minute later is not in it.
+    assert [event["detail"] for event in events] == ["minute-in"]

backend/tests/flow/test_queue.py

@@ -60,6 +60,20 @@ def test_a_deleted_flow_leaves_nothing_parked():
     assert queue.unpark("gone") == []
 
 
+def test_claimed_work_counts_as_in_flight_until_it_is_acknowledged():
+    """The health tile's "in flight" reads zero without this."""
+    queue = MemoryWorkQueue()
+    queue.add(WorkItem(kind="cascade", node="f.n", flow="f"))
+
+    assert queue.stats()["pending"] == 0
+
+    (item,) = queue.claim(1, 10)
+    assert queue.stats()["pending"] == 1
+
+    queue.ack(item)
+    assert queue.stats()["pending"] == 0
+
+
 def _pipeline_with_a_consumer() -> tuple[Pipeline, Node, MemoryState, list]:
     """A source whose message a consumer records."""
     seen: list[float] = []

backend/tests/test_metrics.py

@@ -11,7 +11,7 @@ from datetime import datetime, timezone
 from sqlmodel import Session, select
 
 from app.flow.events import EventBus
-from app.flow.metrics import MetricsCollector
+from app.flow.metrics import DETAIL_CAP, RUN_STALE_S, MetricsCollector
 from app.models import EngineEvent, FlowRun, MetricBucket
 
 FLOW = "metrics-test"
@@ -197,3 +197,30 @@ def test_a_run_that_did_not_fail_reads_ok(db: Session) -> None:
 
     run = db.exec(select(FlowRun).where(FlowRun.id == "manual-abc")).one()
     assert (run.status, run.source) == ("ok", "manual")
+
+
+def test_a_traceback_no_failure_ever_claims_is_dropped() -> None:
+    """One entry per node that ever errored would else last the process out."""
+    node = f"{FLOW}.stale"
+    collector = MetricsCollector(EventBus())
+    collector.handle(
+        {
+            "type": "node_log",
+            "flow": FLOW,
+            "node": node,
+            "level": "error",
+            "text": "x" * (DETAIL_CAP * 2),
+            "ts": time.time() - RUN_STALE_S - 1,
+        }
+    )
+
+    # Untruncated, one of these holds a whole run of a chatty node.
+    assert len(collector._tracebacks[(FLOW, node)][1]) == DETAIL_CAP
+
+    # Something to write, so the flush does not stop at its early return.
+    collector.handle(
+        {"type": "node_executed", "flow": FLOW, "node": node, "ts": time.time()}
+    )
+    asyncio.run(collector.flush())
+
+    assert collector._tracebacks == {}