stroblmeandClaude Fable 5
f300c43f3a
User code no longer execs in the engine. A pool of persistent worker subprocesses speaks one JSON object per line; the controller installs a proxy as the node's function, so every execution path funnels through it and the pipeline is untouched. A crash costs one subprocess, a per-node timeout is a kill, and cancelling from the canvas is that same kill. The workers run a venv of the user's own on the data volume, filled from a pip manifest versioned beside the flows. Applying it retires the workers and rebuilds, so a package lands without restarting the engine. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt
840 lines
30 KiB
Python
840 lines
30 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 traceback
|
|
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 app.core.config import settings
|
|
from app.flow.alerts import AlertManager
|
|
from app.flow.events import EventBus
|
|
from app.flow.executor import ExecutionService
|
|
from app.flow.messages import MessageSpec, flow_of, qualify
|
|
from app.flow.nodes import (
|
|
ChangeNode,
|
|
DelayNode,
|
|
ExecNode,
|
|
FileNode,
|
|
HttpNode,
|
|
InfluxDbNode,
|
|
InjectNode,
|
|
JoinNode,
|
|
MLPNode,
|
|
MqttNode,
|
|
Node,
|
|
NtfyNode,
|
|
RbeNode,
|
|
SwitchNode,
|
|
TriggerNode,
|
|
)
|
|
from app.flow.pipeline import Pipeline, ValidationIssue, ValueSource
|
|
from app.flow.schemas import (
|
|
FlowDef,
|
|
Health,
|
|
NodeDef,
|
|
NodeStatusPublic,
|
|
NodeTypeInfo,
|
|
)
|
|
from app.flow.secrets import SecretNotFound, resolve_params
|
|
from app.flow.state import MemoryState, StateBackend
|
|
from app.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound
|
|
from app.flow.supervision import Supervisor
|
|
from app.flow.worker_main import load_function
|
|
from app.flow.workers import PythonWorkerPool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
HOOK_PREFIX = "/hooks"
|
|
|
|
|
|
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
|
|
|
|
|
|
@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,
|
|
) -> 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
|
|
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()
|
|
self.supervisor = Supervisor(events)
|
|
self.history_limits: dict[str, int] = {}
|
|
self._lock = asyncio.Lock()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
# 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:
|
|
await self._teardown()
|
|
if self.execution is not None:
|
|
await run_in_threadpool(self.execution.stop)
|
|
|
|
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)
|
|
}
|
|
nodes, loaded, initial_values, flow_inputs = self._build_flows(
|
|
[(flow, False) for flow in published]
|
|
)
|
|
|
|
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.
|
|
if entry.flow in self.disabled:
|
|
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]]
|
|
) -> 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)
|
|
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
|
|
) -> 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:
|
|
# 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)
|
|
|
|
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.
|
|
problem = self.workers.compile(owner, local, code)
|
|
if problem:
|
|
entry.status = NodeStatus.ERROR
|
|
entry.error = problem
|
|
return entry
|
|
function = self.workers.proxy(
|
|
owner,
|
|
local,
|
|
code,
|
|
node_id=node_id,
|
|
timeout=node_def.timeout or settings.FLOW_NODE_TIMEOUT,
|
|
)
|
|
node = Node(
|
|
f=function,
|
|
requires=_bound(node_def.requires),
|
|
provides=_bound(node_def.provides),
|
|
params=params,
|
|
name=node_def.id,
|
|
)
|
|
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,
|
|
)
|
|
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) -> str | None:
|
|
"""Does this source load? Returns what to show the author, or None."""
|
|
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 app.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 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 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) -> None:
|
|
"""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.
|
|
"""
|
|
node = self.get_node(node_id)
|
|
if node is None:
|
|
raise KeyError(node_id)
|
|
if node.requires and values:
|
|
node.trigger(values, durable=False)
|
|
else:
|
|
node.inject(values or {}, durable=False)
|
|
|
|
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 _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}"
|