diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py index c60eace..a45c974 100644 --- a/backend/app/api/routes/observability.py +++ b/backend/app/api/routes/observability.py @@ -17,7 +17,7 @@ from sqlalchemy import func from sqlmodel import col, select 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 router = APIRouter( @@ -128,6 +128,22 @@ async def read_summary( if errored: 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 = ( select(func.count()) .select_from(EngineEvent) @@ -146,6 +162,7 @@ async def read_summary( ), "paused": len(paused), "quarantined": len(quarantined), + "invalid": len(invalid), }, nodes={"total": len(entries), "error": len(errored)}, queue=queue, diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 24ff71b..205fe2a 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -62,6 +62,10 @@ logger = logging.getLogger(__name__) 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): ACTIVE = "active" @@ -701,6 +705,15 @@ class FlowController: compiled pipeline, so a node that failed to load still appears — a 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] = {} # Which group each `flow.node_id` ended up in. gid_of: dict[str, str] = {} @@ -738,6 +751,10 @@ class FlowController: group.members.append(member) if flow.name not in group.flows: 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): producers.setdefault(qualify(flow.name, spec.name), []).append( diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index dc4b80b..1a96fba 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -200,6 +200,15 @@ class BrainNode(BaseModel): kind: str members: 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): diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index eff67d4..494541a 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -68,10 +68,56 @@ def test_the_summary_answers_even_when_degraded( assert response.status_code == 200 body = response.json() 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"] +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( client: TestClient, superuser_token_headers: dict[str, str], db: Session ) -> None: diff --git a/backend/tests/flow/test_brain_graph.py b/backend/tests/flow/test_brain_graph.py index 71f44b7..fbd81d0 100644 --- a/backend/tests/flow/test_brain_graph.py +++ b/backend/tests/flow/test_brain_graph.py @@ -12,6 +12,7 @@ import pytest from app.flow.controller import FlowController from app.flow.messages import DType, MessageSpec from app.flow.nodes import MqttNode +from app.flow.pipeline import ValidationIssue from app.flow.schemas import FlowDef, NodeDef 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"}}) 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) diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 5944d47..ee1f1a8 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -265,6 +265,18 @@ export const BrainNodeSchema = { }, type: 'array', 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', diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index d58ddf6..6996b11 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -109,6 +109,10 @@ export type BrainNode = { kind: string; members?: 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); }; /**