"""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"}) # How long a node gets to close what it opened before the rebuild moves on. # A node's `stop` talks to whatever it connected to, and a broker that has gone # away can leave it waiting for an acknowledgement that never arrives — which # used to hold the rebuild, and everything queued behind it, forever. NODE_STOP_TIMEOUT = 5.0 # How long a rebuild asked for by a request waits for one already running. # Generous on purpose: a rebuild of a populated installation reconnects every # node and takes the better part of ten seconds, and a caller queued behind a # healthy one of those should not be turned away. Past that the controller is # wedged rather than busy, and an error the caller can act on beats a request # that never ends. REBUILD_WAIT = 15.0 # How long a caller waits for a *flow* rebuild that is already running. A # per-flow rebuild reconnects one flow's nodes rather than the installation's, # so this is a queueing budget — several of them back to back, which is what # seeding does — not the room a single one needs. Fifteen seconds was sized for # the whole-pipeline rebuild and would let a wedge sit unreported. FLOW_REBUILD_WAIT = 5.0 class RebuildBusy(RuntimeError): """A rebuild could not start because the one before it has not finished. Answered as a 503: nothing is wrong with the request, the engine is busy. """ 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 @dataclass(frozen=True, slots=True) class Wiring: """One published node and the qualified messages it is bound to.""" flow: str node: str type: str provides: list[str] requires: list[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() #: Each flow's declared inputs, kept so rebuilding one flow can #: validate the whole graph without re-reading every flow off the disk. self._flow_inputs: dict[str, dict[str, bool]] = {} self.supervisor = Supervisor(events) self.history_limits: dict[str, int] = {} self._lock = asyncio.Lock() self._failures: asyncio.Task[None] | None = None #: Every published node's ports, with the store revision it was read #: at — see `_wiring`. self._wiring_cache: tuple[int, list[Wiring]] | 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. # Nothing is running to queue behind here, and a lifespan has nobody to # report a timeout to, so this one build waits however long it needs. await self.reload(wait=None) 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, without building anything. Its nodes are built whether or not it runs — being stopped means having no subscriptions, schedules or webhooks, not being absent — so all this has to do is the lifecycle call and the gate that goes with it. Nothing is compiled and no other flow is touched. """ await run_in_threadpool(self.store.write_enabled, flow, enabled) if self.pipeline is None: # Nothing built yet, so there are no nodes to start or stop. await self.reload() return await self._acquire(FLOW_REBUILD_WAIT) try: if enabled: # The gate first: a subscription that fires the instant it # starts must not be turned away by a flag on its way out. self.disabled = self.disabled - {flow} self.pipeline.set_disabled(self.disabled) # Lifts any quarantine, which is what a rebuild used to do by # throwing the whole supervisor away. await self.supervisor.cancel_flow(flow) await self._activate(flow) else: self.disabled = self.disabled | {flow} self.pipeline.set_disabled(self.disabled) await self._teardown(flow) finally: self._lock.release() self._publish( { "type": "pipeline_rebuilt", "issues": [issue.model_dump() for issue in self.issues], "nodes": [status.model_dump() for status in self.node_statuses()], # Every flow's, not this one's: the canvas replaces its paused # set from this, so a narrowed list would clear the markers of # flows this never touched. "paused": self.paused_flows(), } ) async def _acquire(self, wait: float | None) -> None: """Take the rebuild lock, or say the engine is busy. Only one rebuild runs at a time whatever its scope. A caller waits *wait* seconds for the one in front of it and then gives up with ``RebuildBusy`` — hanging on a rebuild that is stuck is worse than saying so. ``None`` waits. """ if wait is None: await self._lock.acquire() return try: await asyncio.wait_for(self._lock.acquire(), wait) except asyncio.TimeoutError: raise RebuildBusy( f"A pipeline rebuild is still running after {wait:.0f}s" ) from None async def reload(self, wait: float | None = REBUILD_WAIT) -> None: """Rebuild the whole pipeline from what is currently stored.""" await self._acquire(wait) try: # 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] ) self._flow_inputs = { flow.name: _declared_inputs(flow)[0] 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) finally: self._lock.release() 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 reload_flow( self, name: str, wait: float | None = FLOW_REBUILD_WAIT ) -> None: """Rebuild one flow, leaving every other flow's nodes connected. A flow is not a subgraph — its nodes can read and write messages another flow owns — but the wiring is derived from message names, so swapping one flow's nodes into the graph and deriving the edges again is enough. What that saves is the reconnecting: the cost of a rebuild on a populated installation is every node opening its socket again, and only one flow's have changed. A flow the store no longer has is taken out instead of replaced. The failure shape is the whole-pipeline one narrowed to a flow, and deliberately no better: a build that raises changes nothing, and anything failing after the teardown leaves that flow's nodes stopped in the graph — which is exactly what a failed ``reload`` leaves behind for all of them. """ pipeline = self.pipeline if pipeline is None: # Nothing built yet, so there is nothing to splice one flow into. await self.reload(wait=wait) return await self._acquire(wait) try: # Built before anything is stopped, so the window in which the # flow is not running is its own teardown and nothing more. Off # the loop for the same reason the full build is: compiling a # python node waits for a worker slot. built = await run_in_threadpool(self._build_one, name) await self._teardown(name) if built is None: pipeline.remove_flow(name) self.loaded = { node_id: entry for node_id, entry in self.loaded.items() if entry.flow != name } # Or the flow would go on declaring inputs nothing provides # any more, and the consumers it left behind would look fine. self._flow_inputs.pop(name, None) self.disabled = self.disabled - {name} self.batch = self.batch - {name} else: flow, nodes, loaded, initial_values, enabled = built pipeline.replace_flow(name, nodes, initial_values) # A rebuild is a fresh set of nodes, but not a fresh history: # a failure nobody has dismissed is the operator's to keep. 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 # Rebound rather than mutated: request threads and the failure # watcher read these without holding anything. self.loaded = { **{ node_id: entry for node_id, entry in self.loaded.items() if entry.flow != name }, **loaded, } self._flow_inputs[name] = _declared_inputs(flow)[0] self.disabled = ( self.disabled - {name} if enabled else self.disabled | {name} ) self.batch = ( self.batch | {name} if flow.mode == "batch" else self.batch - {name} ) pipeline.set_disabled(self.disabled) # Over the whole graph, because it has to be: taking a producer out # of one flow is what leaves another flow's input unconnected. self.issues = _collect_issues( self.loaded, pipeline, self._all_flow_inputs() ) await self._activate(name) # This flow's pause is cleared by the replace, so no resume will # ever come for what it parked. Release it here or it is lost. self._release_parked(name) finally: self._lock.release() self._publish( { "type": "pipeline_rebuilt", "issues": [issue.model_dump() for issue in self.issues], "nodes": [status.model_dump() for status in self.node_statuses()], # Every flow's, not this one's: the canvas replaces its paused # set from this, so a narrowed list would clear the markers of # flows the rebuild never touched. "paused": self.paused_flows(), } ) async def reload_failed_flows(self) -> list[str]: """Rebuild only the flows holding a node that would not load. What installing a module wants: a node red because its import was missing is built again against the packages just installed, and every other flow keeps the connections it already has. """ broken = sorted( { entry.flow for entry in self.loaded.values() if entry.status is NodeStatus.ERROR } ) for flow in broken: await self.reload_flow(flow) return broken async def reload_lib_users(self, ref: str) -> list[str]: """Rebuild the flows running a shared node whose source has changed.""" flows = sorted({usage.split(".", 1)[0] for usage in self.store.usages(ref)}) for flow in flows: await self.reload_flow(flow) return flows def _build_one( self, name: str ) -> tuple[FlowDef, list[Node], dict[str, LoadedNode], dict[str, Any], bool] | None: """Read one published flow and build its nodes. None when it is gone. Blocking: this is the part of a rebuild that talks to the worker pool. """ try: flow = self.store.read_flow(name) except FlowNotFound: return None nodes, loaded, initial_values, _ = self._build_flows([(flow, False)]) return flow, nodes, loaded, initial_values, self.store.read_enabled(name) def _all_flow_inputs(self) -> dict[str, bool]: """Every flow's declared inputs, merged the way one build would see them. In name order, which is the order ``read_all`` returns flows in, so a name two flows both declare resolves to the same one either way. """ merged: dict[str, bool] = {} for name in sorted(self._flow_inputs): merged.update(self._flow_inputs[name]) return merged async def _teardown(self, flow: str | None = None) -> None: """Stop everything the previous pipeline started, or one flow's share.""" for entry in self.loaded.values(): node = entry.node if node is None: continue if flow is not None and entry.flow != flow: continue try: await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT) except asyncio.TimeoutError: # Abandoned rather than waited on: the next node still gets to # close, and the rebuild still happens. logger.warning( "Node '%s' did not stop within %.0fs — carrying on without it", entry.id, NODE_STOP_TIMEOUT, ) except Exception: logger.exception("Error stopping node '%s'", entry.id) # After the nodes, so a loop still winding down is not restarted. if flow is None: await self.supervisor.cancel_all() else: await self.supervisor.cancel_flow(flow) async def _activate(self, flow: str | None = None) -> 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 if flow is not None and entry.flow != flow: 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) declared, initial = _declared_inputs(flow) flow_inputs.update(declared) initial_values.update(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. The source is usually a draft, and the node is usually still running the published one — so ``keep=False``: compile it to answer the question, and leave the worker holding what it is serving calls from. """ 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, keep=False) 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 _wiring(self) -> list[Wiring]: """Every published node and the qualified messages it is bound to. Opening one flow asks what every other flow is wired to, in both directions, which otherwise reads and parses every flow on the disk per request. Rebuilt when the store commits — see `FlowStore.revision` — rather than cached with a lifetime, so a save is visible immediately. """ revision = self.store.revision if self._wiring_cache is not None and self._wiring_cache[0] == revision: return self._wiring_cache[1] index: list[Wiring] = [] for other in self.store.read_all(): for node in other.nodes: provides = sorted( qualify(other.name, spec.name) for spec in _bound(node.provides) ) requires = sorted( qualify(other.name, spec.name) for spec in _bound(node.requires) ) if provides or requires: index.append( Wiring(other.name, node.id, node.type, provides, requires) ) self._wiring_cache = (revision, index) return index 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 entry in self._wiring(): if entry.flow == flow: continue provides = [name for name in entry.provides if name.startswith(prefix)] requires = [name for name in entry.requires if name.startswith(prefix)] if provides or requires: found.append((entry.flow, entry.node, provides, requires)) return found def message_node( self, message: str, published: bool, exclude: str = "" ) -> Wiring | None: """The published node on the far side of ``message``. ``published`` asks for the one that publishes it, otherwise for one that reads it, and ``exclude`` names the flow doing the asking — whose own nodes are never the far side of anything. A message may have several readers; the first stands for them, which is all a label naming the other end has to say. """ for entry in self._wiring(): if entry.flow == exclude: continue if message in (entry.provides if published else entry.requires): return entry return None 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 _declared_inputs(flow: FlowDef) -> tuple[dict[str, bool], dict[str, Any]]: """A flow's declared inputs, and the ones that start with a value.""" declared: dict[str, bool] = {} initial: dict[str, Any] = {} for flow_input in flow.inputs: name = qualify(flow.name, flow_input.spec.name) if not name: continue declared[name] = flow_input.initial is not None if flow_input.initial is not None: initial[name] = flow_input.initial return declared, initial 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("