stroblmeandClaude Opus 5
2c369ac75f
A cluster or GPU host installs `pip install fluksio-worker` and gets the agent and the runner, not psycopg, numpy and the MCP SDK. The engine depends on it as a workspace member, so the file it launches node code with is the same file a remote worker runs — which is what keeps a node unable to tell the difference. Copying the two files by hand still works. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1175 lines
45 KiB
Python
1175 lines
45 KiB
Python
"""Turns stored flows into a running pipeline.
|
|
|
|
The controller is the only thing that builds nodes: it reads flow definitions
|
|
from the store, instantiates each node from its type, and rebuilds the shared
|
|
pipeline. A node that fails to load is reported and skipped — one broken node
|
|
never stops the rest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
import traceback
|
|
from collections.abc import Callable
|
|
from contextlib import suppress
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any, cast
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from fluksio_worker.worker_main import load_function
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow.alerts import AlertManager
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.flow.executor import ExecutionService
|
|
from fluksio.flow.messages import MessageSpec, flow_of, qualify
|
|
from fluksio.flow.nodes import (
|
|
RESERVED_SETTINGS,
|
|
ChangeNode,
|
|
DelayNode,
|
|
ExecNode,
|
|
FileNode,
|
|
HttpNode,
|
|
InfluxDbNode,
|
|
InjectNode,
|
|
JoinNode,
|
|
MLPNode,
|
|
MqttNode,
|
|
Node,
|
|
NtfyNode,
|
|
RbeNode,
|
|
SwitchNode,
|
|
TriggerNode,
|
|
)
|
|
from fluksio.flow.pipeline import NodeOutcome, Pipeline, ValidationIssue, ValueSource
|
|
from fluksio.flow.remote import RemoteWorkerHub
|
|
from fluksio.flow.schemas import (
|
|
BrainEdge,
|
|
BrainGraph,
|
|
BrainNode,
|
|
FlowDef,
|
|
Health,
|
|
NodeDef,
|
|
NodeStatusPublic,
|
|
NodeTypeInfo,
|
|
)
|
|
from fluksio.flow.secrets import SecretNotFound, resolve_params
|
|
from fluksio.flow.state import MemoryState, StateBackend
|
|
from fluksio.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound
|
|
from fluksio.flow.supervision import Supervisor
|
|
from fluksio.flow.workers import PythonWorkerPool
|
|
|
|
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"
|
|
ERROR = "error"
|
|
|
|
|
|
@dataclass
|
|
class LoadedNode:
|
|
"""A node as the editor sees it: built, or failed with a reason.
|
|
|
|
``status`` says whether it loaded; ``health`` says how its connection is
|
|
doing once it is running. A node can be perfectly loaded and unable to
|
|
reach its broker.
|
|
"""
|
|
|
|
id: str
|
|
flow: str
|
|
status: NodeStatus = NodeStatus.ACTIVE
|
|
node: Node | None = None
|
|
error: str | None = None
|
|
health: Health = "ok"
|
|
health_detail: str | None = None
|
|
#: The last time this node raised while running, and what it said. Kept
|
|
#: after it has run again — a failure nobody saw is the one worth keeping —
|
|
#: so only an acknowledgement clears it, not a good run and not a rebuild.
|
|
last_error: str = ""
|
|
last_error_ts: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class RunContext:
|
|
"""The run a set of nodes is being built for.
|
|
|
|
A run builds nodes of its own, so which run a node's worker call belongs to
|
|
is bound when the proxy is made rather than looked up while it runs. That
|
|
is what lets one run's node be cancelled without touching the same node in
|
|
another.
|
|
"""
|
|
|
|
run_id: str
|
|
|
|
|
|
class EmitSink:
|
|
"""Turns a worker's mid-call frames back into the node's own outputs.
|
|
|
|
The proxy has to exist before the node that runs it does, so this stands
|
|
between them: built empty, handed to the proxy, and pointed at the node as
|
|
soon as there is one. What arrives is a dict keyed by output port, which is
|
|
the same thing a return value is — so it goes through the node, gets
|
|
checked against the ports it declared, and is published from there.
|
|
"""
|
|
|
|
__slots__ = ("node",)
|
|
|
|
def __init__(self) -> None:
|
|
self.node: Node | None = None
|
|
|
|
def handle(self, event: dict[str, Any]) -> None:
|
|
if self.node is None or event.get("event") != "emit":
|
|
return
|
|
self.node.emit(event.get("outputs") or {})
|
|
|
|
|
|
@dataclass
|
|
class Preview:
|
|
"""A draft as it would run, without running it."""
|
|
|
|
nodes: list[NodeStatusPublic]
|
|
issues: list[ValidationIssue]
|
|
|
|
|
|
@dataclass
|
|
class NodeType:
|
|
"""A node type the editor can place on a canvas."""
|
|
|
|
title: str
|
|
description: str
|
|
# Constructor signatures differ per node type.
|
|
cls: Any
|
|
has_source: bool = False
|
|
free_params: bool = False
|
|
params_schema: dict[str, Any] = field(default_factory=dict)
|
|
#: Which installed package supplied this type, for the ones that are not
|
|
#: built in.
|
|
plugin: str | None = None
|
|
|
|
|
|
def _schema_of(cls: Any) -> dict[str, Any]:
|
|
params = getattr(cls, "Params", None)
|
|
return params.model_json_schema() if params is not None else {}
|
|
|
|
|
|
NODE_TYPES: dict[str, NodeType] = {
|
|
"python": NodeType(
|
|
title="Function",
|
|
description="Your own Python code, run on every incoming message.",
|
|
cls=Node,
|
|
has_source=True,
|
|
free_params=True,
|
|
),
|
|
"mqtt": NodeType(
|
|
title="MQTT",
|
|
description="Subscribe to topics, or publish what arrives on its inputs.",
|
|
cls=MqttNode,
|
|
params_schema=_schema_of(MqttNode),
|
|
),
|
|
"http": NodeType(
|
|
title="HTTP",
|
|
description="Receive data on a webhook, or send it to a URL.",
|
|
cls=HttpNode,
|
|
params_schema=_schema_of(HttpNode),
|
|
),
|
|
"influxdb": NodeType(
|
|
title="InfluxDB",
|
|
description="Write measurements to a bucket, or read them back.",
|
|
cls=InfluxDbNode,
|
|
params_schema=_schema_of(InfluxDbNode),
|
|
),
|
|
"delay": NodeType(
|
|
title="Delay & schedule",
|
|
description="Hold messages back, limit their rate, or emit on a schedule.",
|
|
cls=DelayNode,
|
|
params_schema=_schema_of(DelayNode),
|
|
),
|
|
"mlp": NodeType(
|
|
title="Perceptron",
|
|
description="A small neural layer over its numeric inputs.",
|
|
cls=MLPNode,
|
|
params_schema=_schema_of(MLPNode),
|
|
),
|
|
"inject": NodeType(
|
|
title="Inject",
|
|
description="Emit a value on request, on a timer, or when the flow starts.",
|
|
cls=InjectNode,
|
|
params_schema=_schema_of(InjectNode),
|
|
),
|
|
"switch": NodeType(
|
|
title="Switch",
|
|
description="Send a value down one branch or another, by rule.",
|
|
cls=SwitchNode,
|
|
params_schema=_schema_of(SwitchNode),
|
|
),
|
|
"change": NodeType(
|
|
title="Change",
|
|
description="Scale, offset, round or map a value on its way past.",
|
|
cls=ChangeNode,
|
|
params_schema=_schema_of(ChangeNode),
|
|
),
|
|
"rbe": NodeType(
|
|
title="Filter unchanged",
|
|
description="Pass a value on only when it has actually changed.",
|
|
cls=RbeNode,
|
|
params_schema=_schema_of(RbeNode),
|
|
),
|
|
"join": NodeType(
|
|
title="Join",
|
|
description="Gather several inputs into one object or list.",
|
|
cls=JoinNode,
|
|
params_schema=_schema_of(JoinNode),
|
|
),
|
|
"trigger": NodeType(
|
|
title="Trigger",
|
|
description="Send one value now and another once things go quiet.",
|
|
cls=TriggerNode,
|
|
params_schema=_schema_of(TriggerNode),
|
|
),
|
|
"exec": NodeType(
|
|
title="Command",
|
|
description="Run a command in the engine's container and read its output.",
|
|
cls=ExecNode,
|
|
params_schema=_schema_of(ExecNode),
|
|
),
|
|
"file": NodeType(
|
|
title="File",
|
|
description="Read a file into the flow, or write one out of it.",
|
|
cls=FileNode,
|
|
params_schema=_schema_of(FileNode),
|
|
),
|
|
"ntfy": NodeType(
|
|
title="Notification",
|
|
description="Push an incoming value to a phone through ntfy.",
|
|
cls=NtfyNode,
|
|
params_schema=_schema_of(NtfyNode),
|
|
),
|
|
}
|
|
|
|
|
|
def node_type_info() -> list[NodeTypeInfo]:
|
|
return [
|
|
NodeTypeInfo(
|
|
type=key,
|
|
title=spec.title,
|
|
description=spec.description,
|
|
params_schema=spec.params_schema,
|
|
has_source=spec.has_source,
|
|
free_params=spec.free_params,
|
|
plugin=spec.plugin,
|
|
)
|
|
for key, spec in NODE_TYPES.items()
|
|
]
|
|
|
|
|
|
class FlowController:
|
|
"""Owns the running pipeline and keeps it in step with the store."""
|
|
|
|
def __init__(
|
|
self,
|
|
store: FlowStore,
|
|
state: StateBackend | None = None,
|
|
events: EventBus | None = None,
|
|
max_workers: int | None = None,
|
|
fastapi_app: FastAPI | None = None,
|
|
execution: ExecutionService | None = None,
|
|
alerts: AlertManager | None = None,
|
|
workers: PythonWorkerPool | None = None,
|
|
remote: RemoteWorkerHub | None = None,
|
|
) -> None:
|
|
self.store = store
|
|
# Without a pool, python nodes are compiled and run in this process —
|
|
# which is what the tests do, and what a bare `Pipeline` has always done.
|
|
self.workers = workers
|
|
# Workers on other hosts. A node without a device never touches it.
|
|
self.remote = remote
|
|
self.state = state if state is not None else MemoryState()
|
|
self.events = events
|
|
self.max_workers = max_workers
|
|
self.app = fastapi_app
|
|
# Without one, every trigger runs inline where it was raised.
|
|
self.execution = execution
|
|
self.alerts = alerts
|
|
# Set by the lifespan once the store exists; the canvas asks the
|
|
# controller which dashboards are wired into a flow.
|
|
self.dashboards: Any = None
|
|
|
|
self.pipeline: Pipeline | None = None
|
|
self.loaded: dict[str, LoadedNode] = {}
|
|
self.issues: list[ValidationIssue] = []
|
|
self.disabled: set[str] = set()
|
|
#: Flows that only run when a run asks them to.
|
|
self.batch: set[str] = set()
|
|
self.supervisor = Supervisor(events)
|
|
self.history_limits: dict[str, int] = {}
|
|
self._lock = asyncio.Lock()
|
|
self._failures: asyncio.Task[None] | None = None
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
# Subscribed before anything runs, so no failure falls between the
|
|
# first build and someone watching for one.
|
|
self._failures = asyncio.create_task(
|
|
self._watch_failures(), name="node-failures"
|
|
)
|
|
# Build first: the consumer must have a pipeline to execute against
|
|
# before it claims anything, or work waiting from the last run would be
|
|
# taken and dropped — which is the very case the queue exists for.
|
|
await self.reload()
|
|
if self.execution is not None:
|
|
self.execution.start()
|
|
|
|
async def stop(self) -> None:
|
|
watcher, self._failures = self._failures, None
|
|
if watcher is not None:
|
|
watcher.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await watcher
|
|
await self._teardown()
|
|
if self.execution is not None:
|
|
await run_in_threadpool(self.execution.stop)
|
|
|
|
async def _watch_failures(self) -> None:
|
|
"""Record every node failure the engine reports, from the event bus.
|
|
|
|
A node can be run by the live pipeline, by a run's own, or by hand, and
|
|
each builds its own graph — but they all report through
|
|
``Pipeline.publish_error``, so the bus is the one place they meet.
|
|
Which is the point: what is recorded here is the node's, not any
|
|
particular pipeline's, and it outlives both the good run that follows
|
|
it and the rebuild after that.
|
|
"""
|
|
if self.events is None:
|
|
return
|
|
async with self.events.subscribe() as queue:
|
|
while True:
|
|
event = await queue.get()
|
|
if event.get("type") != "node_error":
|
|
continue
|
|
entry = self.loaded.get(str(event.get("node") or ""))
|
|
if entry is None:
|
|
continue
|
|
entry.last_error = str(event.get("error") or "")
|
|
entry.last_error_ts = float(event.get("ts") or time.time())
|
|
|
|
def acknowledge_error(self, node_id: str) -> None:
|
|
"""Forget what a node's last failure was. The only thing that clears it.
|
|
|
|
Raises ``KeyError`` for a node the engine does not have, so a route can
|
|
answer that the way every other node call does.
|
|
"""
|
|
entry = self.loaded.get(node_id)
|
|
if entry is None:
|
|
raise KeyError(node_id)
|
|
entry.last_error = ""
|
|
entry.last_error_ts = None
|
|
|
|
async def set_enabled(self, flow: str, enabled: bool) -> None:
|
|
"""Stop or start one flow. Rebuilding is what applies it."""
|
|
await run_in_threadpool(self.store.write_enabled, flow, enabled)
|
|
await self.reload()
|
|
|
|
async def reload(self) -> None:
|
|
"""Rebuild the whole pipeline from what is currently stored."""
|
|
async with self._lock:
|
|
# Work already claimed belongs to the pipeline it was claimed
|
|
# against; let it finish there before swapping the graph out.
|
|
if self.execution is not None:
|
|
await run_in_threadpool(self.execution.pause_intake)
|
|
await self._teardown()
|
|
# A fresh supervisor per build, so a flow quarantined by the last
|
|
# one gets another chance once its author has changed something.
|
|
self.supervisor = Supervisor(self.events)
|
|
|
|
published = self.store.read_all()
|
|
self.disabled = {
|
|
flow.name
|
|
for flow in published
|
|
if not self.store.read_enabled(flow.name)
|
|
}
|
|
# A batch flow is built — the canvas draws it and validation covers
|
|
# it — but never activated: it runs when a run asks it to, and a
|
|
# subscription of its own would be a second way in.
|
|
self.batch = {flow.name for flow in published if flow.mode == "batch"}
|
|
# Off the loop: building a python node asks its worker to compile,
|
|
# which waits for a free slot — and a busy node holds one for as
|
|
# long as its timeout. On the loop that stalls every request, the
|
|
# websocket and the health check the container is judged by.
|
|
nodes, loaded, initial_values, flow_inputs = await run_in_threadpool(
|
|
self._build_flows, [(flow, False) for flow in published]
|
|
)
|
|
|
|
# A rebuild is a fresh set of nodes, but not a fresh history: every
|
|
# publish rebuilds every flow, so dropping the failures here would
|
|
# wipe them constantly. They are the operator's to dismiss.
|
|
for node_id, entry in loaded.items():
|
|
previous = self.loaded.get(node_id)
|
|
if previous is not None and previous.last_error:
|
|
entry.last_error = previous.last_error
|
|
entry.last_error_ts = previous.last_error_ts
|
|
|
|
self.loaded = loaded
|
|
self.pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=self.state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
disabled_flows=self.disabled,
|
|
work_queue=self.execution.queue if self.execution else None,
|
|
node_pool=self.execution.node_pool if self.execution else None,
|
|
)
|
|
self.pipeline.history_limits = self.history_limits
|
|
if self.execution is not None:
|
|
self.execution.bind(self.pipeline)
|
|
self.issues = _collect_issues(loaded, self.pipeline, flow_inputs)
|
|
|
|
await self._activate()
|
|
|
|
if self.execution is not None:
|
|
self.execution.resume_intake()
|
|
# A rebuild clears the pause, so no resume will ever come for
|
|
# what the old pipeline parked. Release it here or it is lost.
|
|
for flow in published:
|
|
self._release_parked(flow.name)
|
|
|
|
self._publish(
|
|
{
|
|
"type": "pipeline_rebuilt",
|
|
"issues": [issue.model_dump() for issue in self.issues],
|
|
"nodes": [status.model_dump() for status in self.node_statuses()],
|
|
# A rebuild is a fresh pipeline, so nothing is paused any more.
|
|
"paused": self.paused_flows(),
|
|
}
|
|
)
|
|
|
|
async def _teardown(self) -> None:
|
|
"""Stop everything the previous pipeline started."""
|
|
for entry in self.loaded.values():
|
|
node = entry.node
|
|
if node is None:
|
|
continue
|
|
try:
|
|
await node.stop(self.app)
|
|
except Exception:
|
|
logger.exception("Error stopping node '%s'", entry.id)
|
|
# After the nodes, so a loop still winding down is not restarted.
|
|
await self.supervisor.cancel_all()
|
|
|
|
async def _activate(self) -> None:
|
|
"""Start subscriptions, schedules and webhooks of the new pipeline.
|
|
|
|
What that means per node type is the node's own business — a connector
|
|
written against the contract starts here the same way the built-ins do.
|
|
"""
|
|
for entry in self.loaded.values():
|
|
node = entry.node
|
|
if node is None:
|
|
continue
|
|
# A stopped flow gets no subscriptions, schedules or webhooks —
|
|
# that is what stopping it means. Nor does a batch flow, which has
|
|
# no outside to listen to.
|
|
if entry.flow in self.disabled or entry.flow in self.batch:
|
|
continue
|
|
node.supervisor = self.supervisor
|
|
try:
|
|
await node.start(self.app)
|
|
except Exception as exc:
|
|
logger.exception("Error starting node '%s'", entry.id)
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = f"{type(exc).__name__}: {exc}"
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Building
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _build_flows(
|
|
self,
|
|
flows: list[tuple[FlowDef, bool]],
|
|
run: RunContext | None = None,
|
|
) -> tuple[list[Node], dict[str, LoadedNode], dict[str, Any], dict[str, bool]]:
|
|
"""Instantiate the nodes of several flows, each published or draft."""
|
|
nodes: list[Node] = []
|
|
loaded: dict[str, LoadedNode] = {}
|
|
initial_values: dict[str, Any] = {}
|
|
# Declared flow inputs, mapped to whether they start with a value.
|
|
flow_inputs: dict[str, bool] = {}
|
|
|
|
for flow, draft in flows:
|
|
for node_def in flow.nodes:
|
|
entry = self._build_node(flow.name, node_def, draft=draft, run=run)
|
|
loaded[entry.id] = entry
|
|
if entry.node is not None:
|
|
nodes.append(entry.node)
|
|
for flow_input in flow.inputs:
|
|
name = qualify(flow.name, flow_input.spec.name)
|
|
if not name:
|
|
continue
|
|
flow_inputs[name] = flow_input.initial is not None
|
|
if flow_input.initial is not None:
|
|
initial_values[name] = flow_input.initial
|
|
|
|
return nodes, loaded, initial_values, flow_inputs
|
|
|
|
def _build_node(
|
|
self,
|
|
flow: str,
|
|
node_def: NodeDef,
|
|
draft: bool = False,
|
|
run: RunContext | None = None,
|
|
) -> LoadedNode:
|
|
node_id = f"{flow}.{node_def.id}"
|
|
entry = LoadedNode(id=node_id, flow=flow)
|
|
try:
|
|
node_type = NODE_TYPES.get(node_def.type)
|
|
if node_type is None:
|
|
raise ValueError(f"Unknown node type '{node_def.type}'")
|
|
|
|
params = resolve_params(node_def.params)
|
|
|
|
if node_type.has_source:
|
|
# Settings and ports are both keyword arguments of the same
|
|
# function, so one name cannot mean both.
|
|
ports = {spec.port for spec in _bound(node_def.requires)}
|
|
clash = sorted((set(params) - RESERVED_SETTINGS) & ports)
|
|
if clash:
|
|
raise ValueError(
|
|
f"'{clash[0]}' is both an input and a setting of "
|
|
f"'{node_def.id}'. A setting is an argument like a "
|
|
"port, so rename one of them."
|
|
)
|
|
|
|
# A shared node runs the library's copy, compiled once under
|
|
# the library's own name so every flow using it agrees.
|
|
if node_def.source_ref:
|
|
owner, local = LIB_DIR, node_def.source_ref
|
|
code = self.store.read_lib_source(node_def.source_ref)
|
|
else:
|
|
owner, local = flow, node_def.id
|
|
code = self.store.read_node_source(flow, node_def.id, draft=draft)
|
|
|
|
# What a node produces before it returns comes back as frames;
|
|
# this puts them through the node's own ports.
|
|
emissions = EmitSink()
|
|
if self.workers is None:
|
|
function = load_function(owner, local, code)
|
|
else:
|
|
# The code never runs here: it is loaded in a worker, and
|
|
# the node calls that worker instead of a local function.
|
|
# A node bound to a device is loaded on *that* machine —
|
|
# one importing torch is correct on the GPU box and a
|
|
# missing module here, so checking it here would fail a
|
|
# node that is fine.
|
|
remote_only = (
|
|
node_def.device
|
|
and node_def.device_policy == "require"
|
|
and self.remote is not None
|
|
)
|
|
if remote_only and self.remote is not None:
|
|
problem = self.remote.compile(
|
|
node_def.device or "", owner, local, code
|
|
)
|
|
else:
|
|
problem = self.workers.compile(owner, local, code)
|
|
if problem:
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = problem
|
|
return entry
|
|
timeout = node_def.timeout or settings.FLOW_NODE_TIMEOUT
|
|
function = self.workers.proxy(
|
|
owner,
|
|
local,
|
|
code,
|
|
node_id=node_id,
|
|
timeout=timeout,
|
|
run_id=run.run_id if run else "",
|
|
on_event=emissions.handle,
|
|
)
|
|
if node_def.device and self.remote is not None:
|
|
# A node with a device runs on a worker carrying that
|
|
# label. Which worker is decided per call, so one that
|
|
# attaches after this flow was built is used without
|
|
# anything being rebuilt.
|
|
function = self.remote.proxy(
|
|
node_def.device,
|
|
owner,
|
|
local,
|
|
code,
|
|
node_id=node_id,
|
|
timeout=timeout,
|
|
run_id=run.run_id if run else "",
|
|
on_event=emissions.handle,
|
|
fallback=(
|
|
function if node_def.device_policy == "prefer" else None
|
|
),
|
|
)
|
|
node = Node(
|
|
f=with_settings(function, params),
|
|
requires=_bound(node_def.requires),
|
|
provides=_bound(node_def.provides),
|
|
params=params,
|
|
name=node_def.id,
|
|
)
|
|
emissions.node = node
|
|
else:
|
|
node = node_type.cls(
|
|
requires=_bound(node_def.requires),
|
|
provides=_bound(node_def.provides),
|
|
params=params,
|
|
name=node_def.id,
|
|
)
|
|
|
|
node.assign_flow(flow, node_def.id)
|
|
node._on_health = self._health_changed
|
|
|
|
if isinstance(node, HttpNode) and node.mode == HttpNode.Mode.TRIGGER:
|
|
# Webhooks live under their flow, away from the JSON API.
|
|
node.url = f"{HOOK_PREFIX}/{flow}/{node.url.lstrip('/')}"
|
|
|
|
entry.node = node
|
|
except (SecretNotFound, LibNotFound) as exc:
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = str(exc)
|
|
except Exception as exc:
|
|
logger.warning("Node '%s' failed to load: %s", node_id, exc)
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = _short_error(exc)
|
|
return entry
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Queries
|
|
# -------------------------------------------------------------------------
|
|
|
|
def node_statuses(self, flow: str | None = None) -> list[NodeStatusPublic]:
|
|
return [
|
|
NodeStatusPublic(
|
|
id=entry.id,
|
|
status=entry.status.value,
|
|
error=entry.error,
|
|
health=entry.health,
|
|
health_detail=entry.health_detail,
|
|
last_error=entry.last_error,
|
|
last_error_ts=entry.last_error_ts,
|
|
)
|
|
for entry in self.loaded.values()
|
|
if flow is None or entry.flow == flow
|
|
]
|
|
|
|
def _health_changed(self, node: Node, status: str, detail: str | None) -> None:
|
|
"""A node saying how its connection is doing, from any thread."""
|
|
entry = self.loaded.get(node.id)
|
|
if entry is None or (entry.health == status and entry.health_detail == detail):
|
|
return
|
|
entry.health = cast(Health, status)
|
|
entry.health_detail = detail
|
|
self._publish(
|
|
{
|
|
"type": "node_health",
|
|
"flow": node.flow,
|
|
"node": node.id,
|
|
"health": status,
|
|
"detail": detail,
|
|
}
|
|
)
|
|
|
|
def flow_issues(self, flow: str) -> list[ValidationIssue]:
|
|
return [issue for issue in self.issues if not issue.flow or issue.flow == flow]
|
|
|
|
def preview(self, name: str) -> Preview:
|
|
"""Build a flow's unpublished draft without deploying it.
|
|
|
|
The draft stands in for its published self among the other flows, so
|
|
cross-flow messages resolve the way they would after publishing. The
|
|
graph is thrown away afterwards and seeded into a throwaway state, so
|
|
nothing here touches what is running.
|
|
"""
|
|
flows: list[tuple[FlowDef, bool]] = [
|
|
(flow, False) for flow in self.store.read_all() if flow.name != name
|
|
]
|
|
try:
|
|
flows.append((self.store.read_flow(name, draft=True), True))
|
|
except FlowNotFound:
|
|
return Preview(nodes=[], issues=[])
|
|
|
|
nodes, loaded, initial_values, flow_inputs = self._build_flows(flows)
|
|
pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=MemoryState(),
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
)
|
|
issues = _collect_issues(loaded, pipeline, flow_inputs)
|
|
return Preview(
|
|
nodes=[
|
|
NodeStatusPublic(
|
|
id=entry.id, status=entry.status.value, error=entry.error
|
|
)
|
|
for entry in loaded.values()
|
|
if entry.flow == name
|
|
],
|
|
issues=[issue for issue in issues if not issue.flow or issue.flow == name],
|
|
)
|
|
|
|
def compile_check(
|
|
self, flow: str, node_id: str, code: str, device: str | None = None
|
|
) -> str | None:
|
|
"""Does this source load? Returns what to show the author, or None.
|
|
|
|
Loaded on the machine that will run it: a node bound to a device is
|
|
checked against that worker's packages, because a missing import here
|
|
says nothing about whether it is missing there.
|
|
"""
|
|
if device and self.remote is not None:
|
|
return self.remote.compile(device, flow, node_id, code)
|
|
if self.workers is not None:
|
|
return self.workers.compile(flow, node_id, code)
|
|
try:
|
|
load_function(flow, node_id, code)
|
|
except Exception as exc:
|
|
return _short_error(exc)
|
|
return None
|
|
|
|
def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
|
|
return self.pipeline.values(flow) if self.pipeline else {}
|
|
|
|
def get_node(self, node_id: str) -> Node | None:
|
|
entry = self.loaded.get(node_id)
|
|
return entry.node if entry else None
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Execution
|
|
# -------------------------------------------------------------------------
|
|
|
|
def is_enabled(self, flow: str) -> bool:
|
|
return flow not in self.disabled
|
|
|
|
def is_paused(self, flow: str) -> bool:
|
|
return self.pipeline is not None and flow in self.pipeline.paused_flows()
|
|
|
|
def is_quarantined(self, flow: str) -> bool:
|
|
return flow in self.supervisor.quarantined
|
|
|
|
@property
|
|
def quarantined(self) -> set[str]:
|
|
return self.supervisor.quarantined
|
|
|
|
def paused_flows(self) -> list[str]:
|
|
return self.pipeline.paused_flows() if self.pipeline else []
|
|
|
|
def pause_flow(self, flow: str) -> None:
|
|
if self.pipeline is not None:
|
|
self.pipeline.pause(flow)
|
|
|
|
def resume_flow(self, flow: str) -> None:
|
|
"""Blocking — call from a worker thread: held-back nodes run on resume."""
|
|
if self.pipeline is None:
|
|
return
|
|
self.pipeline.resume(flow)
|
|
self._release_parked(flow)
|
|
|
|
def _release_parked(self, flow: str) -> None:
|
|
"""Queue what a pause held back again, oldest first, so it is not lost."""
|
|
if self.execution is None:
|
|
return
|
|
for item in self.execution.queue.unpark(flow):
|
|
self.execution.queue.add(item)
|
|
|
|
def step_flow(self, flow: str) -> str | None:
|
|
"""Run one held-back item, leaving the flow paused. Blocking.
|
|
|
|
Returns the node it came from, or None when nothing is held back.
|
|
"""
|
|
return self.execution.step(flow) if self.execution is not None else None
|
|
|
|
def set_history_limits(self, limits: dict[str, int]) -> None:
|
|
"""How deep to keep each charted message's series. Applies at once."""
|
|
self.history_limits = limits
|
|
if self.pipeline is not None:
|
|
self.pipeline.history_limits = limits
|
|
|
|
def message_catalog(self) -> list[Any]:
|
|
"""Every message the published flows declare, with its last value.
|
|
|
|
What a dashboard picks from, so it spans flows rather than sitting
|
|
inside one.
|
|
"""
|
|
from fluksio.api.routes.messages import MessageInfo
|
|
|
|
specs: dict[str, MessageSpec] = {}
|
|
providers: dict[str, list[str]] = {}
|
|
for flow in self.store.read_all():
|
|
for node in flow.nodes:
|
|
for spec in _bound(node.provides):
|
|
name = qualify(flow.name, spec.name)
|
|
specs.setdefault(name, spec)
|
|
providers.setdefault(name, []).append(f"{flow.name}.{node.id}")
|
|
for spec in _bound(node.requires):
|
|
specs.setdefault(qualify(flow.name, spec.name), spec)
|
|
for declared in flow.inputs:
|
|
if declared.spec.name:
|
|
specs.setdefault(
|
|
qualify(flow.name, declared.spec.name), declared.spec
|
|
)
|
|
|
|
values = self.values()
|
|
infos = []
|
|
for name, spec in sorted(specs.items()):
|
|
current = values.get(name, {})
|
|
infos.append(
|
|
MessageInfo(
|
|
name=name,
|
|
flow=flow_of(name),
|
|
dtype=spec.dtype.value,
|
|
providers=sorted(providers.get(name, [])),
|
|
writable=True,
|
|
numeric=spec.dtype.value in ("float", "int"),
|
|
value=current.get("value"),
|
|
ts=current.get("ts"),
|
|
)
|
|
)
|
|
return infos
|
|
|
|
def cross_flow_nodes(
|
|
self, flow: str
|
|
) -> list[tuple[str, str, list[str], list[str]]]:
|
|
"""Nodes in other flows on the far side of a dotted message name.
|
|
|
|
Returns ``(flow, node id, messages of `flow` it provides, ones it
|
|
requires)``. This is what link-in and link-out mean: the wiring is
|
|
already real — the canvas simply had no way to draw the other end.
|
|
"""
|
|
prefix = f"{flow}."
|
|
found = []
|
|
for other in self.store.read_all():
|
|
if other.name == flow:
|
|
continue
|
|
for node in other.nodes:
|
|
provides = sorted(
|
|
qualify(other.name, spec.name)
|
|
for spec in _bound(node.provides)
|
|
if qualify(other.name, spec.name).startswith(prefix)
|
|
)
|
|
requires = sorted(
|
|
qualify(other.name, spec.name)
|
|
for spec in _bound(node.requires)
|
|
if qualify(other.name, spec.name).startswith(prefix)
|
|
)
|
|
if provides or requires:
|
|
found.append((other.name, node.id, provides, requires))
|
|
return found
|
|
|
|
def brain_graph(self) -> BrainGraph:
|
|
"""Every published flow as one graph, merged on what its nodes talk to.
|
|
|
|
Two flows on the same broker topic are one neuron here: the topic is a
|
|
single thing, and the path that runs through it is real wiring that no
|
|
one flow's canvas can show. Read from the store rather than from the
|
|
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] = {}
|
|
producers: dict[str, list[str]] = {}
|
|
# (member, message) for every input that actually wakes a node.
|
|
consumers: list[tuple[str, str]] = []
|
|
|
|
for flow in self.store.read_all():
|
|
for node_def in flow.nodes:
|
|
member = f"{flow.name}.{node_def.id}"
|
|
node_type = NODE_TYPES.get(node_def.type)
|
|
key = None
|
|
if node_type is not None:
|
|
try:
|
|
key = node_type.cls.instance_key(node_def.params)
|
|
except Exception:
|
|
# A plugin's own grouping is not worth the whole graph;
|
|
# this node just stands on its own.
|
|
logger.exception(
|
|
"instance_key failed for node type '%s'", node_def.type
|
|
)
|
|
gid = f"{node_def.type}:{key}" if key else member
|
|
gid_of[member] = gid
|
|
|
|
group = groups.get(gid)
|
|
if group is None:
|
|
# The first member names the neuron. What it points at is
|
|
# in the id, which can be a whole topic map and is far too
|
|
# long to draw on a circle.
|
|
group = groups[gid] = BrainNode(
|
|
id=gid,
|
|
label=node_def.title or node_def.id,
|
|
kind=node_def.type,
|
|
)
|
|
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(
|
|
member
|
|
)
|
|
for spec in _bound(node_def.requires):
|
|
if spec.trigger:
|
|
consumers.append((member, qualify(flow.name, spec.name)))
|
|
|
|
messages: dict[tuple[str, str], set[str]] = {}
|
|
for member, message in consumers:
|
|
target = gid_of[member]
|
|
for producer in producers.get(message, ()):
|
|
source = gid_of[producer]
|
|
# Everything a merged neuron says to itself is inside it.
|
|
if source != target:
|
|
messages.setdefault((source, target), set()).add(message)
|
|
|
|
return BrainGraph(
|
|
nodes=sorted(groups.values(), key=lambda node: node.id),
|
|
edges=[
|
|
BrainEdge(source=source, target=target, messages=sorted(names))
|
|
for (source, target), names in sorted(messages.items())
|
|
],
|
|
)
|
|
|
|
def publish_message(
|
|
self, name: str, value: Any, source: ValueSource | None = None
|
|
) -> None:
|
|
"""Put a value into the graph from outside. Blocking.
|
|
|
|
Refuses a name no flow declares: the flows own the namespace, and a
|
|
message nothing reads or writes would just be a key nobody sees.
|
|
"""
|
|
if self.pipeline is None:
|
|
raise KeyError(name)
|
|
|
|
spec = None
|
|
for flow in self.store.read_all():
|
|
for node in flow.nodes:
|
|
for candidate in [*_bound(node.provides), *_bound(node.requires)]:
|
|
if qualify(flow.name, candidate.name) == name:
|
|
spec = candidate
|
|
break
|
|
for declared in flow.inputs:
|
|
if qualify(flow.name, declared.spec.name) == name:
|
|
spec = declared.spec
|
|
if spec is None:
|
|
raise KeyError(name)
|
|
|
|
# The same check a node's output gets; a dashboard is not looser.
|
|
spec.check(value)
|
|
self.pipeline.publish({name: value}, source)
|
|
|
|
def queue_stats(self) -> dict[str, Any]:
|
|
return self.execution.stats() if self.execution is not None else {}
|
|
|
|
def forget_flow(self, flow: str) -> None:
|
|
"""Drop what a deleted flow left behind. Blocking."""
|
|
if self.execution is not None:
|
|
self.execution.queue.clear_flow(flow)
|
|
prefix = f"{flow}."
|
|
with self.state.lock():
|
|
stale = [
|
|
key
|
|
for key in self.state.keys()
|
|
# Both the messages themselves and the engine's own bookkeeping
|
|
# about them, which is keyed by message name too.
|
|
if key.startswith(prefix) or f":{prefix}" in key
|
|
]
|
|
for key in stale:
|
|
self.state.delete(key)
|
|
|
|
def run_flow(self, flow: str, inputs: dict[str, Any] | None = None) -> None:
|
|
"""Run every node of one flow. Blocking — call from a worker thread."""
|
|
if self.pipeline is None:
|
|
return
|
|
self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
|
|
|
|
def build_run_pipeline(
|
|
self,
|
|
flow: FlowDef,
|
|
state: StateBackend,
|
|
draft: bool = False,
|
|
observer: Callable[[NodeOutcome], None] | None = None,
|
|
emission_observer: Callable[[str, dict[str, Any]], None] | None = None,
|
|
run: RunContext | None = None,
|
|
) -> Pipeline:
|
|
"""Build one flow as a pipeline of its own, for a single run.
|
|
|
|
Nothing here is activated: a run executes the graph from its inputs
|
|
rather than waiting to be told something, so subscriptions, schedules
|
|
and webhooks would only be a second copy of what the live pipeline
|
|
already holds. The state is the run's, which is what keeps two runs of
|
|
one flow from overwriting each other's messages.
|
|
"""
|
|
nodes, _loaded, initial_values, _inputs = self._build_flows(
|
|
[(flow, draft)], run=run
|
|
)
|
|
pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
observer=observer,
|
|
emission_observer=emission_observer,
|
|
)
|
|
pipeline.history_limits = self.history_limits
|
|
return pipeline
|
|
|
|
def run_preview(self, flow_name: str, inputs: dict[str, Any] | None = None) -> None:
|
|
"""Run a flow's unpublished draft once, as the editor shows it.
|
|
|
|
The nodes are built and run but never activated, so a draft never
|
|
acquires subscriptions, schedules or webhooks — only publishing does
|
|
that. State and events are the live ones, so values land on the canvas
|
|
exactly as they do for a published run.
|
|
"""
|
|
flow = self.store.read_flow(flow_name, draft=True)
|
|
nodes, _, initial_values, _ = self._build_flows([(flow, True)])
|
|
pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=self.state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
)
|
|
pipeline.run(inputs or {})
|
|
|
|
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)
|
|
try:
|
|
if node.requires and values:
|
|
node.trigger(values, 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:
|
|
self.events.publish(event)
|
|
|
|
|
|
def _bound(specs: list[MessageSpec]) -> list[MessageSpec]:
|
|
"""Ports without a message name are not wired into the graph."""
|
|
return [spec for spec in specs if spec.name]
|
|
|
|
|
|
def with_settings(
|
|
function: Callable[..., Any], params: dict[str, Any]
|
|
) -> Callable[..., Any]:
|
|
"""A node's function with its settings bound as keyword arguments.
|
|
|
|
A setting is a constant of one node's function, so it is passed the way a
|
|
port is: by name. The engine's own settings never reach the code, and the
|
|
``params`` the pipeline offers is dropped here rather than travelling to a
|
|
worker that has nothing to do with it.
|
|
"""
|
|
settings = {k: v for k, v in params.items() if k not in RESERVED_SETTINGS}
|
|
|
|
def call(
|
|
params: dict[str, Any] | None = None, # noqa: ARG001 - absorbed here
|
|
**ports: Any,
|
|
) -> Any:
|
|
return function(**ports, **settings)
|
|
|
|
return call
|
|
|
|
|
|
def _collect_issues(
|
|
loaded: dict[str, LoadedNode],
|
|
pipeline: Pipeline,
|
|
flow_inputs: dict[str, bool],
|
|
) -> list[ValidationIssue]:
|
|
"""Everything wrong with a built graph, whether it is running or not."""
|
|
issues = pipeline.validate(flow_inputs)
|
|
issues += [
|
|
ValidationIssue(
|
|
code="node_error",
|
|
message=entry.error or "This node could not be loaded.",
|
|
flow=entry.flow,
|
|
node=entry.id,
|
|
)
|
|
for entry in loaded.values()
|
|
if entry.status is NodeStatus.ERROR
|
|
]
|
|
# Webhooks are mounted without authentication, so a hook without a secret is
|
|
# open to anyone who guesses its URL. Flows written before the parameter
|
|
# existed keep running, but say so.
|
|
issues += [
|
|
ValidationIssue(
|
|
code="unauthenticated_hook",
|
|
message=(
|
|
f"Webhook '{entry.node.local_id}' has no secret — "
|
|
"anyone who knows its URL can trigger it."
|
|
),
|
|
flow=entry.flow,
|
|
node=entry.id,
|
|
)
|
|
for entry in loaded.values()
|
|
if isinstance(entry.node, HttpNode)
|
|
and entry.node.mode == HttpNode.Mode.TRIGGER
|
|
and not entry.node.secret
|
|
]
|
|
return issues
|
|
|
|
|
|
def _short_error(exc: Exception) -> str:
|
|
"""One line a node author can act on: what broke, and where in their code.
|
|
|
|
The frames in between belong to the loader rather than to the node, so only
|
|
the line in the node's own source is kept. The full traceback stays in the
|
|
server log.
|
|
"""
|
|
if isinstance(exc, SyntaxError):
|
|
# Its own message already names the compiled file, which is noise here.
|
|
return f"{type(exc).__name__}: {exc.msg} (line {exc.lineno})"
|
|
|
|
frames = [
|
|
frame
|
|
for frame in traceback.extract_tb(exc.__traceback__)
|
|
if frame.filename.startswith("<node ")
|
|
]
|
|
where = f" (line {frames[-1].lineno})" if frames else ""
|
|
return f"{type(exc).__name__}: {exc}{where}"
|