Health: a flow that cannot run says so, and the brain marks which neurons
A dependency loop is flagged on the canvas and was invisible everywhere else: /observability/summary answered "ok" with an empty problems list while the published flow could not run at all. It now reports the flows validation blocks, and the brain graph carries the reason on each neuron the issue names so the view built to find broken wiring can show it. Node errors stay counted once, as the nodes that failed to load, and an advisory like an unauthenticated webhook marks nothing — it is worth saying, but the flow still runs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XC2jX6Hdj7pxGGKzBTrbqB
This commit is contained in:
@@ -17,7 +17,7 @@ from sqlalchemy import func
|
|||||||
from sqlmodel import col, select
|
from sqlmodel import col, select
|
||||||
|
|
||||||
from app.api.deps import FlowControllerDep, SessionDep, get_current_user
|
from app.api.deps import FlowControllerDep, SessionDep, get_current_user
|
||||||
from app.flow.controller import NodeStatus
|
from app.flow.controller import ADVISORY_ISSUES, NodeStatus
|
||||||
from app.models import EngineEvent, FlowRun, MetricBucket
|
from app.models import EngineEvent, FlowRun, MetricBucket
|
||||||
|
|
||||||
router = APIRouter(
|
router = APIRouter(
|
||||||
@@ -128,6 +128,22 @@ async def read_summary(
|
|||||||
if errored:
|
if errored:
|
||||||
problems.append(f"{len(errored)} node(s) failed to load")
|
problems.append(f"{len(errored)} node(s) failed to load")
|
||||||
|
|
||||||
|
# What the canvas flags on a flow — a dependency loop, an input nothing
|
||||||
|
# feeds — stops that flow running just as surely as a node that will not
|
||||||
|
# load, and until now this screen was the one place it did not show.
|
||||||
|
# `node_error` is left out: those are the nodes already counted above.
|
||||||
|
invalid = sorted(
|
||||||
|
{
|
||||||
|
issue.flow
|
||||||
|
for issue in controller.issues
|
||||||
|
if issue.flow
|
||||||
|
and issue.code != "node_error"
|
||||||
|
and issue.code not in ADVISORY_ISSUES
|
||||||
|
}
|
||||||
|
)
|
||||||
|
if invalid:
|
||||||
|
problems.append(f"{len(invalid)} flow(s) cannot run: {', '.join(invalid)}")
|
||||||
|
|
||||||
statement = (
|
statement = (
|
||||||
select(func.count())
|
select(func.count())
|
||||||
.select_from(EngineEvent)
|
.select_from(EngineEvent)
|
||||||
@@ -146,6 +162,7 @@ async def read_summary(
|
|||||||
),
|
),
|
||||||
"paused": len(paused),
|
"paused": len(paused),
|
||||||
"quarantined": len(quarantined),
|
"quarantined": len(quarantined),
|
||||||
|
"invalid": len(invalid),
|
||||||
},
|
},
|
||||||
nodes={"total": len(entries), "error": len(errored)},
|
nodes={"total": len(entries), "error": len(errored)},
|
||||||
queue=queue,
|
queue=queue,
|
||||||
|
|||||||
@@ -62,6 +62,10 @@ logger = logging.getLogger(__name__)
|
|||||||
|
|
||||||
HOOK_PREFIX = "/hooks"
|
HOOK_PREFIX = "/hooks"
|
||||||
|
|
||||||
|
# Validation codes that are worth saying but do not stop a flow running, so
|
||||||
|
# neither the brain graph nor the health summary treats them as a fault.
|
||||||
|
ADVISORY_ISSUES = frozenset({"unauthenticated_hook"})
|
||||||
|
|
||||||
|
|
||||||
class NodeStatus(str, Enum):
|
class NodeStatus(str, Enum):
|
||||||
ACTIVE = "active"
|
ACTIVE = "active"
|
||||||
@@ -701,6 +705,15 @@ class FlowController:
|
|||||||
compiled pipeline, so a node that failed to load still appears — a
|
compiled pipeline, so a node that failed to load still appears — a
|
||||||
broken neuron is exactly what someone comes to this view to find.
|
broken neuron is exactly what someone comes to this view to find.
|
||||||
"""
|
"""
|
||||||
|
# What validation found, by the node it names. A cycle names every node
|
||||||
|
# in it, so all of them are marked rather than an arbitrary one.
|
||||||
|
troubled: dict[str, str] = {}
|
||||||
|
for issue in self.issues:
|
||||||
|
if issue.code in ADVISORY_ISSUES:
|
||||||
|
continue
|
||||||
|
for member in (*issue.nodes, *filter(None, [issue.node])):
|
||||||
|
troubled.setdefault(member, issue.message)
|
||||||
|
|
||||||
groups: dict[str, BrainNode] = {}
|
groups: dict[str, BrainNode] = {}
|
||||||
# Which group each `flow.node_id` ended up in.
|
# Which group each `flow.node_id` ended up in.
|
||||||
gid_of: dict[str, str] = {}
|
gid_of: dict[str, str] = {}
|
||||||
@@ -738,6 +751,10 @@ class FlowController:
|
|||||||
group.members.append(member)
|
group.members.append(member)
|
||||||
if flow.name not in group.flows:
|
if flow.name not in group.flows:
|
||||||
group.flows.append(flow.name)
|
group.flows.append(flow.name)
|
||||||
|
# A merged neuron stands for several nodes, so one of them being
|
||||||
|
# unable to run is enough to mark it; the first reason wins.
|
||||||
|
if group.issue is None:
|
||||||
|
group.issue = troubled.get(member)
|
||||||
|
|
||||||
for spec in _bound(node_def.provides):
|
for spec in _bound(node_def.provides):
|
||||||
producers.setdefault(qualify(flow.name, spec.name), []).append(
|
producers.setdefault(qualify(flow.name, spec.name), []).append(
|
||||||
|
|||||||
@@ -200,6 +200,15 @@ class BrainNode(BaseModel):
|
|||||||
kind: str
|
kind: str
|
||||||
members: list[str] = Field(default_factory=list)
|
members: list[str] = Field(default_factory=list)
|
||||||
flows: list[str] = Field(default_factory=list)
|
flows: list[str] = Field(default_factory=list)
|
||||||
|
issue: str | None = Field(
|
||||||
|
default=None,
|
||||||
|
description=(
|
||||||
|
"Why this neuron cannot run, if validation found something. A "
|
||||||
|
"failure the engine hits while running arrives over the socket "
|
||||||
|
"instead; this is the part that is already true before anything "
|
||||||
|
"fires, and so has to travel with the graph."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class BrainEdge(BaseModel):
|
class BrainEdge(BaseModel):
|
||||||
|
|||||||
@@ -68,10 +68,56 @@ def test_the_summary_answers_even_when_degraded(
|
|||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
body = response.json()
|
body = response.json()
|
||||||
assert body["status"] in {"ok", "degraded"}
|
assert body["status"] in {"ok", "degraded"}
|
||||||
assert set(body["flows"]) == {"total", "running", "paused", "quarantined"}
|
assert set(body["flows"]) == {
|
||||||
|
"total",
|
||||||
|
"running",
|
||||||
|
"paused",
|
||||||
|
"quarantined",
|
||||||
|
"invalid",
|
||||||
|
}
|
||||||
assert "error" in body["nodes"]
|
assert "error" in body["nodes"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_flow_that_cannot_run_makes_the_summary_degraded(
|
||||||
|
client: TestClient, superuser_token_headers: dict[str, str]
|
||||||
|
) -> None:
|
||||||
|
"""A loop on the canvas has to reach the health screen.
|
||||||
|
|
||||||
|
Validation runs on a build, not on this request, so what the engine already
|
||||||
|
knows is what this reports — and reporting it is the whole point: a flow
|
||||||
|
with a dependency loop cannot run, and the screen used to say "ok".
|
||||||
|
"""
|
||||||
|
from app.flow.pipeline import ValidationIssue
|
||||||
|
|
||||||
|
controller = client.app.state.flow_controller
|
||||||
|
before = controller.issues
|
||||||
|
controller.issues = [
|
||||||
|
ValidationIssue(
|
||||||
|
code="cycle",
|
||||||
|
message="These nodes depend on each other in a loop",
|
||||||
|
flow="looping",
|
||||||
|
nodes=["looping.a", "looping.b"],
|
||||||
|
),
|
||||||
|
# Already counted as a node that failed to load, so not again here.
|
||||||
|
ValidationIssue(
|
||||||
|
code="node_error", message="boom", flow="broken", node="broken.x"
|
||||||
|
),
|
||||||
|
ValidationIssue(
|
||||||
|
code="unauthenticated_hook", message="open", flow="hooky", node="hooky.h"
|
||||||
|
),
|
||||||
|
]
|
||||||
|
try:
|
||||||
|
body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json()
|
||||||
|
finally:
|
||||||
|
controller.issues = before
|
||||||
|
|
||||||
|
assert body["status"] == "degraded"
|
||||||
|
assert body["flows"]["invalid"] == 1
|
||||||
|
assert any("looping" in problem for problem in body["problems"])
|
||||||
|
assert not any("broken" in problem for problem in body["problems"])
|
||||||
|
assert not any("hooky" in problem for problem in body["problems"])
|
||||||
|
|
||||||
|
|
||||||
def test_the_history_reads_back(
|
def test_the_history_reads_back(
|
||||||
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
client: TestClient, superuser_token_headers: dict[str, str], db: Session
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import pytest
|
|||||||
from app.flow.controller import FlowController
|
from app.flow.controller import FlowController
|
||||||
from app.flow.messages import DType, MessageSpec
|
from app.flow.messages import DType, MessageSpec
|
||||||
from app.flow.nodes import MqttNode
|
from app.flow.nodes import MqttNode
|
||||||
|
from app.flow.pipeline import ValidationIssue
|
||||||
from app.flow.schemas import FlowDef, NodeDef
|
from app.flow.schemas import FlowDef, NodeDef
|
||||||
from app.flow.store import FlowStore
|
from app.flow.store import FlowStore
|
||||||
|
|
||||||
@@ -80,3 +81,38 @@ def test_a_credential_never_reaches_the_key():
|
|||||||
key = MqttNode.instance_key({**BROKER, "password": {"$secret": "broker_pw"}})
|
key = MqttNode.instance_key({**BROKER, "password": {"$secret": "broker_pw"}})
|
||||||
|
|
||||||
assert key == "mosquitto:1883/sensors/temp"
|
assert key == "mosquitto:1883/sensors/temp"
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_neuron_carries_what_stops_it_running(controller: FlowController):
|
||||||
|
# Validation runs on a build, so the graph on its own knows nothing yet.
|
||||||
|
assert all(node.issue is None for node in controller.brain_graph().nodes)
|
||||||
|
|
||||||
|
controller.issues = [
|
||||||
|
ValidationIssue(
|
||||||
|
code="cycle",
|
||||||
|
message="These nodes depend on each other in a loop",
|
||||||
|
flow="house",
|
||||||
|
nodes=["house.scale", "house.sensor"],
|
||||||
|
)
|
||||||
|
]
|
||||||
|
graph = controller.brain_graph()
|
||||||
|
|
||||||
|
# Both named nodes are marked, the merged neuron among them, and the flow
|
||||||
|
# that has nothing wrong with it is left alone.
|
||||||
|
assert {node.id: node.issue is not None for node in graph.nodes} == {
|
||||||
|
"house.scale": True,
|
||||||
|
"mqtt:mosquitto:1883/sensors/temp": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_an_advisory_issue_leaves_the_graph_clean(controller: FlowController):
|
||||||
|
controller.issues = [
|
||||||
|
ValidationIssue(
|
||||||
|
code="unauthenticated_hook",
|
||||||
|
message="Webhook 'hook' has no secret",
|
||||||
|
flow="house",
|
||||||
|
node="house.scale",
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
assert all(node.issue is None for node in controller.brain_graph().nodes)
|
||||||
|
|||||||
@@ -265,6 +265,18 @@ export const BrainNodeSchema = {
|
|||||||
},
|
},
|
||||||
type: 'array',
|
type: 'array',
|
||||||
title: 'Flows'
|
title: 'Flows'
|
||||||
|
},
|
||||||
|
issue: {
|
||||||
|
anyOf: [
|
||||||
|
{
|
||||||
|
type: 'string'
|
||||||
|
},
|
||||||
|
{
|
||||||
|
type: 'null'
|
||||||
|
}
|
||||||
|
],
|
||||||
|
title: 'Issue',
|
||||||
|
description: 'Why this neuron cannot run, if validation found something. A failure the engine hits while running arrives over the socket instead; this is the part that is already true before anything fires, and so has to travel with the graph.'
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
type: 'object',
|
type: 'object',
|
||||||
|
|||||||
@@ -109,6 +109,10 @@ export type BrainNode = {
|
|||||||
kind: string;
|
kind: string;
|
||||||
members?: Array<(string)>;
|
members?: Array<(string)>;
|
||||||
flows?: Array<(string)>;
|
flows?: Array<(string)>;
|
||||||
|
/**
|
||||||
|
* Why this neuron cannot run, if validation found something. A failure the engine hits while running arrives over the socket instead; this is the part that is already true before anything fires, and so has to travel with the graph.
|
||||||
|
*/
|
||||||
|
issue?: (string | null);
|
||||||
};
|
};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
Reference in New Issue
Block a user