Melvin StroblandClaude Fable 5
8c82549cf6
The browser half of M3. Flows open on a full-bleed canvas with their chrome floating over it: flow tabs top, dock bottom, node settings in a panel on the right that leaves the graph visible and running behind it. - Connections are derived, not stored. A node declares the messages it reads and publishes; every matching pair draws an edge, so two producers of one message converge on their consumer. Dragging output to input is shorthand for pointing that input at the producer's message, and asks before it replaces an existing one. - Values land on the edges as they flow, over a websocket that feeds a store outside React, so a value arriving re-renders its own chip and nothing else. Clicking an edge shows the last payload and when it arrived. - Node source is edited in Monaco, loaded only when a panel opens and themed from the design tokens. - Edits autosave; identical documents are skipped server-side, so a quiet canvas writes nothing. - Validation from the API shows on the node it belongs to and is summarised in the dock, where each entry pans to its node. - Works on a phone: touch-connect, 44px dock targets, and the node panel becomes a full-screen sheet. Two new tokens (--status-success, --font-mono) are mirrored in the website repo and recorded in DESIGN-GUIDELINES.md. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016WzrvW7rjQbynnhF6pxh6i
394 lines
13 KiB
Python
394 lines
13 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 hashlib
|
|
import logging
|
|
import sys
|
|
import traceback
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from types import ModuleType
|
|
from typing import Any, cast
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from app.flow.events import EventBus
|
|
from app.flow.messages import MessageSpec, qualify
|
|
from app.flow.nodes import (
|
|
DelayNode,
|
|
HttpNode,
|
|
InfluxDbNode,
|
|
MLPNode,
|
|
MqttNode,
|
|
Node,
|
|
)
|
|
from app.flow.pipeline import Pipeline, ValidationIssue
|
|
from app.flow.schemas import NodeDef, NodeStatusPublic, NodeTypeInfo
|
|
from app.flow.secrets import SecretNotFound, resolve_params
|
|
from app.flow.state import MemoryState, StateBackend
|
|
from app.flow.store import FlowStore
|
|
|
|
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."""
|
|
|
|
id: str
|
|
flow: str
|
|
status: NodeStatus = NodeStatus.ACTIVE
|
|
node: Node | None = None
|
|
error: str | None = None
|
|
|
|
|
|
@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
|
|
params_schema: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
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,
|
|
),
|
|
"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),
|
|
),
|
|
}
|
|
|
|
|
|
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,
|
|
)
|
|
for key, spec in NODE_TYPES.items()
|
|
]
|
|
|
|
|
|
def _load_function(flow: str, node_id: str, code: str) -> Callable[..., Any]:
|
|
"""Compile a node's source and return the function to run.
|
|
|
|
A node file defines ``process(...)``; if it defines exactly one public
|
|
function under another name, that one is used.
|
|
"""
|
|
digest = hashlib.md5(code.encode()).hexdigest()[:8]
|
|
module_name = f"_fluksio_node_{flow}_{node_id}_{digest}"
|
|
|
|
module = ModuleType(module_name)
|
|
module.__dict__["__name__"] = module_name
|
|
sys.modules[module_name] = module
|
|
exec(compile(code, f"<node {flow}.{node_id}>", "exec"), module.__dict__)
|
|
|
|
if callable(getattr(module, "process", None)):
|
|
return cast(Callable[..., Any], module.process)
|
|
|
|
functions = [
|
|
value
|
|
for name, value in vars(module).items()
|
|
if callable(value)
|
|
and not name.startswith("_")
|
|
and getattr(value, "__module__", None) == module_name
|
|
]
|
|
if len(functions) == 1:
|
|
return cast(Callable[..., Any], functions[0])
|
|
raise ValueError(
|
|
"Define a function named 'process' — this file has "
|
|
f"{len(functions)} functions to choose from."
|
|
)
|
|
|
|
|
|
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,
|
|
) -> None:
|
|
self.store = store
|
|
self.state = state or MemoryState()
|
|
self.events = events
|
|
self.max_workers = max_workers
|
|
self.app = fastapi_app
|
|
|
|
self.pipeline: Pipeline | None = None
|
|
self.loaded: dict[str, LoadedNode] = {}
|
|
self.issues: list[ValidationIssue] = []
|
|
self._lock = asyncio.Lock()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
await self.reload()
|
|
|
|
async def stop(self) -> None:
|
|
await self._teardown()
|
|
|
|
async def reload(self) -> None:
|
|
"""Rebuild the whole pipeline from what is currently stored."""
|
|
async with self._lock:
|
|
await self._teardown()
|
|
|
|
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 in self.store.read_all():
|
|
for node_def in flow.nodes:
|
|
entry = self._build_node(flow.name, node_def)
|
|
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
|
|
|
|
self.loaded = loaded
|
|
self.pipeline = Pipeline(
|
|
nodes=nodes,
|
|
state=self.state,
|
|
events=self.events,
|
|
max_workers=self.max_workers,
|
|
initial_values=initial_values,
|
|
)
|
|
self.issues = self.pipeline.validate(flow_inputs)
|
|
self.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
|
|
]
|
|
|
|
await self._activate()
|
|
|
|
self._publish(
|
|
{
|
|
"type": "pipeline_rebuilt",
|
|
"issues": [issue.model_dump() for issue in self.issues],
|
|
"nodes": [status.model_dump() for status in self.node_statuses()],
|
|
}
|
|
)
|
|
|
|
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:
|
|
if isinstance(node, MqttNode) and node.is_subscribed:
|
|
await node.stop_subscription()
|
|
elif isinstance(node, DelayNode) and node.cron_expr:
|
|
await node.stop_cron()
|
|
if (
|
|
isinstance(node, HttpNode)
|
|
and node.mode == HttpNode.Mode.TRIGGER
|
|
and self.app is not None
|
|
):
|
|
node.unregister_route(self.app)
|
|
except Exception:
|
|
logger.exception("Error stopping node '%s'", entry.id)
|
|
|
|
async def _activate(self) -> None:
|
|
"""Start subscriptions, schedules and webhooks of the new pipeline."""
|
|
for entry in self.loaded.values():
|
|
node = entry.node
|
|
if node is None:
|
|
continue
|
|
try:
|
|
if isinstance(node, MqttNode) and node.mode == MqttNode.Mode.SUBSCRIBER:
|
|
await node.start_subscription()
|
|
elif isinstance(node, DelayNode) and node.cron_expr:
|
|
await node.start_cron()
|
|
elif (
|
|
isinstance(node, HttpNode)
|
|
and node.mode == HttpNode.Mode.TRIGGER
|
|
and self.app is not None
|
|
):
|
|
node.register_route(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_node(self, flow: str, node_def: NodeDef) -> 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:
|
|
code = self.store.read_node_source(flow, node_def.id)
|
|
node = Node(
|
|
f=_load_function(flow, node_def.id, code),
|
|
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)
|
|
|
|
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 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_traceback(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)
|
|
for entry in self.loaded.values()
|
|
if flow is None or entry.flow == flow
|
|
]
|
|
|
|
def flow_issues(self, flow: str) -> list[ValidationIssue]:
|
|
return [issue for issue in self.issues if not issue.flow or issue.flow == flow]
|
|
|
|
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 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 trigger_node(self, node_id: str, values: dict[str, Any] | None = None) -> None:
|
|
"""Feed values into one node. Blocking — call from a worker thread."""
|
|
node = self.get_node(node_id)
|
|
if node is None:
|
|
raise KeyError(node_id)
|
|
if node.requires and values:
|
|
node.trigger(values)
|
|
else:
|
|
node.inject(values or {})
|
|
|
|
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 _short_traceback(exc: Exception) -> str:
|
|
"""The last frames of a failure, which is what a node author needs."""
|
|
lines = traceback.format_exception(type(exc), exc, exc.__traceback__)
|
|
return "".join(lines[-3:]).strip()
|