Refuse a port the function cannot take, and read a failing node as degraded

A python node's ports and settings arrive as keyword arguments, so a declared
name its `process` does not take was a TypeError on every call — and a node
that loads fine and fails every time it runs is the quiet kind of broken: the
hosted demo did it 720 times an hour for two days and the health badge read
ok throughout. `_build_node` now reads a written body with `ast` and refuses
the mismatch at load, so the node is an error on the canvas and an issue on
publish. Skipped for `**kwargs`, a decorated or absent `process`, and the
template a new node opens with. The SDK's generated shim always takes
`**settings`, so synced flows are untouched.

`/observability/summary` names a node that has failed in the last fifteen
minutes and reads degraded while it does, which is what would have made the
badge amber. `nodes.failing` carries the count.

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

backend/fluksio/api/routes/observability.py

@@ -7,6 +7,7 @@ which the generated SDK turns into a thrown error — and a health page that
 cannot render while the engine is degraded is the wrong way round.
 """
 
+import time
 from datetime import UTC, datetime, timedelta
 from typing import Annotated, Any, Literal
 
@@ -191,6 +192,20 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
             f"{len(unhealthy)} node(s) down: "
             f"{', '.join(sorted(e.id for e in unhealthy))}"
         )
+    # A node that loads and then fails on every call is not in `errored`, and
+    # until this it read as healthy. ponytail: one failure reads degraded for
+    # 15 minutes; a counter with decay if that proves noisy.
+    now = time.time()
+    failing = [
+        e
+        for e in entries
+        if e.last_error_ts is not None and now - e.last_error_ts < 900
+    ]
+    if failing:
+        problems.append(
+            f"{len(failing)} node(s) failing: "
+            f"{', '.join(sorted(e.id for e in failing))}"
+        )
 
     # 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
@@ -234,6 +249,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any:
             "total": len(entries),
             "error": len(errored),
             "unhealthy": len(unhealthy),
+            "failing": len(failing),
         },
         queue=queue,
         loop_lag=(

backend/fluksio/flow/controller.py

@@ -14,6 +14,7 @@ nodes are built like any other's, so `set_enabled` only starts or stops them.
 
 from __future__ import annotations
 
+import ast
 import asyncio
 import hashlib
 import json
@@ -1027,6 +1028,19 @@ class FlowController:
                         flow, node_def.id
                     )
 
+                # Only a body somebody wrote: the template a new node opens
+                # with takes nothing, and a node declared before it is written
+                # is reported as missing its source rather than as wrong.
+                written = bool(node_def.source_ref) or self.store.has_node_source(
+                    flow, node_def.id, draft=draft
+                )
+                if written:
+                    gap = _signature_gap(
+                        code, node_def.id, ports, set(params) - RESERVED_SETTINGS
+                    )
+                    if gap:
+                        raise ValueError(gap)
+
                 # What a node produces before it returns comes back as frames;
                 # this puts them through the node's own ports.
                 emissions = EmitSink()
@@ -1716,6 +1730,42 @@ def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
     return [spec for spec in specs if spec.name]
 
 
+def _signature_gap(
+    code: str, node_id: str, ports: set[str], settings: set[str]
+) -> str | None:
+    """A declared port or setting ``process`` cannot accept, or None.
+
+    Ports and settings both arrive as keyword arguments, so a name the function
+    does not take is a ``TypeError`` on every call — and a node that loads fine
+    and fails every time it runs is the quiet kind of broken. Read statically:
+    with a worker pool the code is never imported here. Anything this cannot
+    read for certain (no plain ``def process``, a decorator, ``**kwargs``, a
+    syntax error the compile step words better) is left to the call.
+    """
+    try:
+        tree = ast.parse(code)
+    except SyntaxError:
+        return None
+    defs = [
+        node
+        for node in tree.body
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef)
+        and node.name == "process"
+    ]
+    if not defs or defs[-1].decorator_list or defs[-1].args.kwarg is not None:
+        return None
+    args = defs[-1].args
+    names = [a.arg for a in [*args.posonlyargs, *args.args, *args.kwonlyargs]]
+    missing = sorted((ports | settings) - set(names))
+    if not missing:
+        return None
+    kind = "an input" if missing[0] in ports else "a setting"
+    return (
+        f"'{missing[0]}' is {kind} of '{node_id}' but "
+        f"process({', '.join(names)}) takes no such argument"
+    )
+
+
 def with_settings(
     function: Callable[..., Any], params: dict[str, Any]
 ) -> Callable[..., Any]:

backend/tests/api/routes/test_observability.py

@@ -447,3 +447,38 @@ def test_a_timeseries_window_can_be_named(
     assert response.status_code == 200
     # Two of the four minutes: `since` inclusive, `until` exclusive.
     assert sum(point["executions"] for point in response.json()) == 2
+
+
+def test_a_node_failing_on_every_call_makes_the_summary_degraded(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """A node that loads and then raises each time it runs is not in `error`.
+
+    It is named while its last failure is recent, and drops out once it is not.
+    """
+    import time
+
+    from fluksio.flow.controller import LoadedNode
+
+    controller = client.app.state.flow_controller
+    before = controller.loaded
+    controller.loaded = {
+        "house.calc": LoadedNode(
+            id="house.calc", flow="house", last_error="boom", last_error_ts=time.time()
+        ),
+        "house.old": LoadedNode(
+            id="house.old",
+            flow="house",
+            last_error="boom",
+            last_error_ts=time.time() - 3600,
+        ),
+    }
+    try:
+        body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json()
+    finally:
+        controller.loaded = before
+
+    assert body["status"] == "degraded"
+    assert body["nodes"]["failing"] == 1
+    (problem,) = [p for p in body["problems"] if "failing" in p]
+    assert "house.calc" in problem and "house.old" not in problem

backend/tests/flow/test_node_settings.py

@@ -61,3 +61,90 @@ def test_a_setting_named_after_a_port_is_refused(store: FlowStore):
 
     assert [node.status for node in preview.nodes] == ["error"]
     assert "both an input and a setting" in (preview.nodes[0].error or "")
+
+
+def _flow(requires: list[MessageSpec], **params: object) -> FlowDef:
+    return FlowDef(
+        name="house",
+        nodes=[
+            NodeDef(
+                id="scale",
+                params=dict(params),
+                requires=requires,
+                provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
+            )
+        ],
+    )
+
+
+READING = [MessageSpec(name="reading", dtype=DType.FLOAT)]
+
+
+def _preview(store: FlowStore, flow: FlowDef, source: str | None):
+    store.write_draft(flow, 0)
+    if source is not None:
+        store.write_node_source("house", "scale", source, draft=True)
+    return FlowController(store).preview("house").nodes[0]
+
+
+def test_a_port_the_function_cannot_take_is_refused_at_load(store: FlowStore):
+    node = _preview(store, _flow(READING), "def process(value):\n    return {}\n")
+
+    assert node.status == "error"
+    assert "'reading' is an input of 'scale'" in (node.error or "")
+    assert "process(value)" in (node.error or "")
+
+
+def test_a_setting_the_function_cannot_take_is_refused_at_load(store: FlowStore):
+    node = _preview(store, _flow(READING, gain=2), SOURCE)
+
+    assert node.status == "error"
+    assert "'gain' is a setting of 'scale'" in (node.error or "")
+
+
+def test_a_function_taking_keywords_accepts_anything(store: FlowStore):
+    node = _preview(
+        store, _flow(READING, gain=2), "def process(**kw):\n    return {}\n"
+    )
+
+    assert node.status == "active"
+
+
+def test_a_node_never_written_is_not_wrong_yet(store: FlowStore):
+    # The template a new node opens with takes nothing; that is missing, not a
+    # mismatch.
+    node = _preview(store, _flow(READING), None)
+
+    assert node.status == "active"
+
+
+def test_a_port_arrives_under_its_own_name(store: FlowStore):
+    other = [MessageSpec(name="other.reading", dtype=DType.FLOAT)]
+    node = _preview(store, _flow(other, factor=3), SOURCE)
+
+    assert node.status == "active"
+
+
+def test_a_shared_body_is_checked_against_each_user(store: FlowStore):
+    store.write_flow(_flow(READING, factor=3))
+    store.write_node_source("house", "scale", SOURCE)
+    store.share_node("house", "scale", "scale_it")
+    store.write_flow(
+        FlowDef(
+            name="garden",
+            nodes=[
+                NodeDef(
+                    id="scale",
+                    source_ref="scale_it",
+                    requires=[MessageSpec(name="level", dtype=DType.FLOAT)],
+                    provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
+                )
+            ],
+        )
+    )
+    controller = FlowController(store)
+
+    nodes = controller.preview("garden").nodes
+
+    assert [n.status for n in nodes] == ["error"]
+    assert "'level' is an input of 'scale'" in (nodes[0].error or "")