From 3724b68f230c5e5160bc9c0cb91e020e5ebbf295 Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Sat, 15 Aug 2026 23:57:44 +0200 Subject: [PATCH] Add the connector contract, reusable nodes and per-port intervals Connectors are the device-facing node class third parties write, so the surface they build against is versioned and documented: ConnectorNode carries a declared contract version, a polling loop that publishes only what changed and reports health around it, and parameters whose credential fields are marked x-secret so the editor offers the secrets store instead of a text box. They are found through the fluksio.node_types entry point group, with the package's own metadata as the manifest. docs/connectors/ has the contract and the authoring guide; connector-skeleton/ is a working one to copy. The controller no longer knows what any node type is: start, stop and report_health are protocol methods on Node, and the built-ins were migrated to them first, so the hooks a connector implements are the ones the engine has been driving all along. Marking a node reusable moves its source to _lib/ and points the node at it by name. Other flows instantiate it with their own ports and settings, one fix reaches all of them, and a shared source still in use cannot be deleted. Ports gained an interval: an output publishes, and an input wakes its node, at most every n seconds. State keeps the latest value, so only the delivery is skipped, and pressing Run is never throttled. Also fixes autosave sending no version on its first save of a session, which made every flow saved more than once conflict with itself. Co-Authored-By: Claude Fable 5 --- NOTEPAD.md | 11 +- ROADMAP.md | 21 +- backend/app/api/routes/flows.py | 120 ++++++++++- backend/app/flow/connector.py | 154 +++++++++++++++ backend/app/flow/controller.py | 90 ++++++--- backend/app/flow/messages.py | 7 +- backend/app/flow/nodes.py | 59 +++++- backend/app/flow/pipeline.py | 72 ++++++- backend/app/flow/plugins.py | 77 ++++++++ backend/app/flow/schemas.py | 21 +- backend/app/flow/store.py | 124 ++++++++++++ backend/app/main.py | 3 + backend/tests/flow/test_connector.py | 156 +++++++++++++++ backend/tests/flow/test_discretization.py | 100 ++++++++++ backend/tests/flow/test_library.py | 96 +++++++++ frontend/src/client/schemas.gen.ts | 96 ++++++++- frontend/src/client/sdk.gen.ts | 89 ++++++++- frontend/src/client/types.gen.ts | 49 ++++- .../src/components/Flow/CommandPalette.tsx | 32 +++ frontend/src/components/Flow/FlowEditor.tsx | 11 +- frontend/src/components/Flow/NodePanel.tsx | 186 +++++++++++++++++- frontend/src/components/Flow/queries.ts | 29 ++- 22 files changed, 1541 insertions(+), 62 deletions(-) create mode 100644 backend/app/flow/connector.py create mode 100644 backend/app/flow/plugins.py create mode 100644 backend/tests/flow/test_connector.py create mode 100644 backend/tests/flow/test_discretization.py create mode 100644 backend/tests/flow/test_library.py diff --git a/NOTEPAD.md b/NOTEPAD.md index 432f865..e44bd92 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -27,9 +27,14 @@ Deferring because out of scope is fine, but don't mention deferring than. - FEAT/UI: publishing and discarding are only reachable while no side panel is open, since the floating chrome hides for the panel. Editing a node's code and publishing it is therefore close-panel-then-publish. -- FEAT/UI: no way to point a node parameter at a stored secret. The store and the - `{"$secret": "name"}` reference both work, but the params form only produces literals, so a - webhook secret typed there lands in `flow.json`. +- FEAT/UI: there is no screen for managing the secrets store itself. A node parameter marked + `x-secret` offers the stored secrets, but they can only be created through the API. +- FEAT/FLOW: input discretization drops the trailing edge — if a producer goes quiet inside + the window, the consumer never sees the last value. Delivering it at the end of the window + needs a timer per (node, port). +- CHORE/FLOW: shared node sources bypass the draft/publish split. Editing one writes the + library copy and reloads immediately, since the code is not any single flow's to hold back. + Deliberate, but it means a shared node is the one thing publish does not gate. - FEAT/UI: flow-level edits (title, inputs) go through `FlowPanel.onChange` and stay outside the undo stack. - CHORE/UI: `make test-backend` cannot reach Postgres while the integrated stack is up — diff --git a/ROADMAP.md b/ROADMAP.md index d981963..ce90f50 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -42,11 +42,15 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M - [x] Secrets/credentials store for node integrations managed via the API/UI (encrypted at rest, referenced from node params as `{"$secret": "name"}`); `.env` bootstrap-only -- [ ] Connector node contract: declared I/O schemas, UI-driven credential and - setup flow, polling/discretization behaviour, health reporting, packaging - metadata — plus an authoring guide and skeleton repo. Device-specific - nodes are written against it as external plugins; the registry follows - later +- [x] Connector node contract: `ConnectorNode` with a declared contract version, + a polling coordinator that deduplicates, `x-secret` parameters the editor + renders as a secret picker, and health reporting. Connectors are installed + packages found through the `fluksio.node_types` entry point group; the + contract is documented in `docs/connectors/` with a working skeleton at + `connector-skeleton/`. The registry follows later +- [x] Node lifecycle as a protocol (`start`/`stop`/`report_health` on `Node`), + replacing the controller's per-type isinstance chains — the same hooks a + connector implements, validated on the built-in nodes first - [x] Flow persistence: `flow.json` plus node sources per flow, replacing the watch-directory prototype - [x] REST + WebSocket API over the engine: create/read/update flows, edit node @@ -64,7 +68,9 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M draft. Saves carry the version they were based on, so a second client editing the same flow is refused rather than overwritten - [ ] Import/export of a flow as human-readable code plus a JSON structure -- [ ] Per-input/-output discretization interval setting +- [x] Per-input/-output discretization interval setting: a port publishes, or + wakes its node, at most every n seconds. State keeps the latest value, so + only the delivery is skipped - [ ] Alert / notification handler - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking deployment on failure @@ -99,6 +105,9 @@ React + Vite, primarily desktop but usable on mobile. See `docs/architecture/str - [x] Validation shown on the node it belongs to, and summarised in the dock - [x] Publish control and draft markers in the flow bar, discard in the flow panel, and a conflict dialog when another client got there first +- [x] Marking a node reusable, and placing a shared one from the palette +- [x] Secret picker for credential parameters, so a password never lands in + `flow.json` - [x] Dashboard showing which flows run, which are stopped and which have errors, with a switch per flow - [x] Logs panel in the canvas dock, pause/resume beside Run, and replaying an diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 664c81b..e008108 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -27,6 +27,7 @@ from app.flow.schemas import ( FlowStatePublic, FlowSummary, HistoryPoint, + LibraryNode, MessageHistory, MessageValue, NodeSource, @@ -34,7 +35,13 @@ from app.flow.schemas import ( NodeTypeInfo, ) from app.flow.state import as_number -from app.flow.store import FlowExists, FlowNotFound, StaleVersion +from app.flow.store import ( + FlowExists, + FlowNotFound, + LibExists, + LibNotFound, + StaleVersion, +) from app.models import Message router = APIRouter( @@ -73,6 +80,10 @@ class PublishRequest(BaseModel): version: int +class ShareRequest(BaseModel): + lib_name: str + + class RunRequest(BaseModel): inputs: dict[str, Any] = {} @@ -114,6 +125,12 @@ def _read_flow(controller: FlowController, name: str) -> FlowDef: raise HTTPException(status_code=404, detail=f"No flow named '{name}'") +def _source_ref(definition: FlowDef, node_id: str) -> str | None: + """The library source this node runs, if it is a shared one.""" + node = next((n for n in definition.nodes if n.id == node_id), None) + return node.source_ref if node else None + + def _require_enabled(controller: FlowController, name: str) -> None: if not controller.is_enabled(name): raise HTTPException( @@ -168,6 +185,39 @@ def read_node_types() -> Any: return node_type_info() +# ----------------------------------------------------------------------------- +# Shared nodes +# +# Declared above the "/{name}" routes: "library" would otherwise be read as a +# flow name. +# ----------------------------------------------------------------------------- + + +@router.get("/library", response_model=list[LibraryNode]) +def read_library(controller: FlowControllerDep) -> Any: + """The node sources shared across flows, and which nodes use each.""" + return [ + LibraryNode(name=name, used_by=controller.store.usages(name)) + for name in controller.store.list_lib() + ] + + +@router.delete("/library/{lib_name}", response_model=Message) +async def delete_shared_node(lib_name: str, controller: FlowControllerDep) -> Any: + """Remove a shared source, as long as no flow still runs it.""" + used_by = controller.store.usages(lib_name) + if used_by: + raise HTTPException( + status_code=409, + detail=f"'{lib_name}' is still used by {', '.join(used_by)}", + ) + try: + await run_in_threadpool(controller.store.delete_lib_source, lib_name) + except LibNotFound: + raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'") + return Message(message=f"Deleted shared node '{lib_name}'") + + @router.get("/{name}", response_model=FlowDetail) def read_flow(name: str, controller: FlowControllerDep) -> Any: """Read one flow, with the state of its nodes.""" @@ -303,7 +353,13 @@ def read_node_source( controller: FlowControllerDep, ) -> Any: """Read a node's Python source, including unpublished edits.""" - _read_flow(controller, name) + definition = _read_flow(controller, name) + ref = _source_ref(definition, node_id) + if ref: + try: + return NodeSource(code=controller.store.read_lib_source(ref)) + except LibNotFound: + raise HTTPException(status_code=404, detail=f"No shared node '{ref}'") return NodeSource(code=controller.store.read_node_source(name, node_id, draft=True)) @@ -319,11 +375,23 @@ async def save_node_source( The answer comes from compiling the code rather than from the running pipeline: a draft is not deployed, and compiling is both faster and more precise about what the author just typed. + + A shared node writes to the library, so the fix reaches every flow using + it — and that one is live immediately rather than waiting for a publish, + because the copy is not any single flow's to hold back. """ - _read_flow(controller, name) - await run_in_threadpool( - controller.store.write_node_source, name, node_id, source.code, True - ) + definition = _read_flow(controller, name) + ref = _source_ref(definition, node_id) + if ref: + changed = await run_in_threadpool( + controller.store.write_lib_source, ref, source.code + ) + if changed: + await controller.reload() + else: + await run_in_threadpool( + controller.store.write_node_source, name, node_id, source.code, True + ) error = await run_in_threadpool( controller.compile_check, name, node_id, source.code ) @@ -334,6 +402,46 @@ async def save_node_source( ) +@router.post("/{name}/nodes/{node_id}/share", response_model=FlowDetail) +async def share_node( + name: str, + node_id: str, + body: ShareRequest, + controller: FlowControllerDep, +) -> Any: + """Move this node's code into the library so other flows can run it too.""" + _read_flow(controller, name) + if not NAME_PATTERN.match(body.lib_name): + raise HTTPException( + status_code=400, + detail=( + "Use lowercase letters, digits and underscores, starting with a letter" + ), + ) + try: + await run_in_threadpool( + controller.store.share_node, name, node_id, body.lib_name + ) + except LibExists as exc: + raise HTTPException(status_code=409, detail=str(exc)) + return _detail(controller, _read_flow(controller, name)) + + +@router.post("/{name}/nodes/{node_id}/unshare", response_model=FlowDetail) +async def unshare_node( + name: str, + node_id: str, + controller: FlowControllerDep, +) -> Any: + """Take a private copy of the shared code back into this flow.""" + _read_flow(controller, name) + try: + await run_in_threadpool(controller.store.unshare_node, name, node_id) + except LibNotFound as exc: + raise HTTPException(status_code=404, detail=str(exc)) + return _detail(controller, _read_flow(controller, name)) + + # ----------------------------------------------------------------------------- # Running, stopped, paused # ----------------------------------------------------------------------------- diff --git a/backend/app/flow/connector.py b/backend/app/flow/connector.py new file mode 100644 index 0000000..a570719 --- /dev/null +++ b/backend/app/flow/connector.py @@ -0,0 +1,154 @@ +"""The contract a connector node is written against. + +A connector is the device-facing node class: it talks to something outside the +engine — a device, a service, a protocol — and publishes what it finds as +ordinary typed messages. Third parties write these, so this surface is the one +part of the engine that has to stay stable; it is versioned by +:data:`CONTRACT_VERSION` and a connector declares which version it was written +for. + +What a connector gets from the base class: + +* a polling loop that runs :meth:`ConnectorNode.poll` on a schedule, publishes + only the ports whose value changed, and reports health around it; +* :meth:`Node.report_health`, so a connection problem shows on the node rather + than only in the log; +* the lifecycle hooks the controller drives, so nothing device-specific has to + be known by the engine. + +The message schemas and the parameter model are the rest of the contract, and +they are the same ones the built-in nodes use. See ``docs/connectors/`` for the +authoring guide. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import TYPE_CHECKING, Any, ClassVar + +from pydantic import BaseModel, Field + +from app.flow.nodes import Node + +if TYPE_CHECKING: + from fastapi import FastAPI + +logger = logging.getLogger(__name__) + +#: Bumped when a change would break connectors written against the old surface. +#: The loader refuses a connector declaring anything else. +CONTRACT_VERSION = 1 + + +class ConnectorNode(Node): + """Base class for device- and service-facing nodes. + + A subclass declares the contract version it was written for, describes + itself for the editor, and implements :meth:`poll`, :meth:`start`, or both:: + + class RandomSensor(ConnectorNode): + contract = CONTRACT_VERSION + title = "Random sensor" + description = "Emits a random reading, for trying the contract out." + + class Params(ConnectorNode.Params): + ceiling: float = 1.0 + + async def poll(self): + return {"reading": random.random() * self.config.ceiling} + """ + + #: Declared explicitly by every connector; inheriting it does not count. + contract: ClassVar[int] + title: ClassVar[str] = "" + description: ClassVar[str] = "" + + class Params(BaseModel): + """Settings the editor renders a form for. + + Subclass it to add your own. A field holding a credential should carry + ``json_schema_extra={"x-secret": True}``, which makes the editor offer + the stored secrets instead of a text box. + """ + + poll_interval: float = Field( + default=0, + ge=0, + description="Seconds between polls; 0 polls never.", + ) + + __slots__ = ("config", "_poll_task", "_stop_event", "_last_published") + + def __init__(self, **kwargs: Any) -> None: + super().__init__(f=self._unused, **kwargs) + self.config = type(self).Params(**self.params) + self._poll_task: asyncio.Task[None] | None = None + self._stop_event: asyncio.Event | None = None + self._last_published: dict[str, Any] = {} + + @staticmethod + def _unused(**_: Any) -> None: + """A connector publishes from its own loop, not from the scheduler.""" + return None + + # ------------------------------------------------------------------------- + # What a connector implements + # ------------------------------------------------------------------------- + + async def poll(self) -> dict[str, Any] | None: + """Read the device once and return values keyed by output port. + + Return ``None`` when there is nothing new. Raising is reported as a + health problem and retried on the next tick. + """ + return None + + # ------------------------------------------------------------------------- + # What the engine drives + # ------------------------------------------------------------------------- + + async def start(self, app: FastAPI | None = None) -> None: + if self.config.poll_interval > 0 and self._poll_task is None: + self._stop_event = asyncio.Event() + self._poll_task = asyncio.create_task(self._poll_loop()) + + async def stop(self, app: FastAPI | None = None) -> None: + if self._poll_task is None: + return + if self._stop_event is not None: + self._stop_event.set() + self._poll_task.cancel() + try: + await self._poll_task + except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down + pass + self._poll_task = None + self._last_published = {} + + async def _poll_loop(self) -> None: + """Poll, publish what changed, and say how the connection is doing. + + Only changed ports are published: a device polled every few seconds is + usually saying the same thing, and every publication wakes everything + downstream of it. + """ + while not (self._stop_event and self._stop_event.is_set()): + try: + values = await self.poll() + self.report_health("ok") + changed = { + port: value + for port, value in (values or {}).items() + if self._last_published.get(port, object()) != value + } + if changed: + self._last_published.update(changed) + # inject runs the graph, which is blocking work. + await asyncio.to_thread(self.inject, changed) + except asyncio.CancelledError: + break + except Exception as exc: + logger.warning("Connector '%s' failed to poll: %s", self.id, exc) + self.report_health("down", f"{type(exc).__name__}: {exc}") + await asyncio.sleep(self.config.poll_interval) diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index 6c930f6..71e7818 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -33,10 +33,16 @@ from app.flow.nodes import ( Node, ) from app.flow.pipeline import Pipeline, ValidationIssue -from app.flow.schemas import FlowDef, NodeDef, NodeStatusPublic, NodeTypeInfo +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 FlowNotFound, FlowStore +from app.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound logger = logging.getLogger(__name__) @@ -50,13 +56,20 @@ class NodeStatus(str, Enum): @dataclass class LoadedNode: - """A node as the editor sees it: built, or failed with a reason.""" + """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 @@ -77,6 +90,9 @@ class NodeType: cls: Any has_source: 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]: @@ -132,6 +148,7 @@ def node_type_info() -> list[NodeTypeInfo]: description=spec.description, params_schema=spec.params_schema, has_source=spec.has_source, + plugin=spec.plugin, ) for key, spec in NODE_TYPES.items() ] @@ -252,21 +269,16 @@ class FlowController: 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) + await node.stop(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.""" + """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: @@ -276,16 +288,7 @@ class FlowController: if entry.flow in self.disabled: 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) + await node.start(self.app) except Exception as exc: logger.exception("Error starting node '%s'", entry.id) entry.status = NodeStatus.ERROR @@ -334,9 +337,16 @@ class FlowController: params = resolve_params(node_def.params) if node_type.has_source: - code = self.store.read_node_source(flow, node_def.id, draft=draft) + # 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: + code = self.store.read_lib_source(node_def.source_ref) + function = _load_function(LIB_DIR, node_def.source_ref, code) + else: + code = self.store.read_node_source(flow, node_def.id, draft=draft) + function = _load_function(flow, node_def.id, code) node = Node( - f=_load_function(flow, node_def.id, code), + f=function, requires=_bound(node_def.requires), provides=_bound(node_def.provides), params=params, @@ -351,13 +361,14 @@ class FlowController: ) 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 as exc: + except (SecretNotFound, LibNotFound) as exc: entry.status = NodeStatus.ERROR entry.error = str(exc) except Exception as exc: @@ -372,11 +383,34 @@ class FlowController: def node_statuses(self, flow: str | None = None) -> list[NodeStatusPublic]: return [ - NodeStatusPublic(id=entry.id, status=entry.status.value, error=entry.error) + 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] diff --git a/backend/app/flow/messages.py b/backend/app/flow/messages.py index 1947ebd..f0ffc08 100644 --- a/backend/app/flow/messages.py +++ b/backend/app/flow/messages.py @@ -14,7 +14,7 @@ import json from enum import Enum from typing import Any -from pydantic import BaseModel, ConfigDict, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_validator class DType(str, Enum): @@ -42,6 +42,10 @@ class MessageSpec(BaseModel): :param port: The identifier the node function sees. Defaults to the last segment of ``name``, so unqualified flows read naturally. :param dtype: Payload type, validated on every message that passes through. + :param interval: Deliver at most every this many seconds; 0 is every time. + On an output it holds back publishing, on an input it holds back waking + the node. The value is never lost — state keeps the latest — only the + delivery is skipped. """ model_config = ConfigDict(frozen=True) @@ -49,6 +53,7 @@ class MessageSpec(BaseModel): name: str = "" port: str = "" dtype: DType = DType.FLOAT + interval: float = Field(default=0, ge=0) @model_validator(mode="after") def _default_port(self) -> MessageSpec: diff --git a/backend/app/flow/nodes.py b/backend/app/flow/nodes.py index 1140ee4..1e03f2f 100644 --- a/backend/app/flow/nodes.py +++ b/backend/app/flow/nodes.py @@ -74,6 +74,7 @@ class Node: "params", "_pipeline", "synchronous", + "_on_health", ) def __init__( @@ -86,6 +87,7 @@ class Node: ): self.f = f self._pipeline: Pipeline | None = None + self._on_health: Callable[[Node, str, str | None], None] | None = None self.params = dict(params) if params else {} self.synchronous = bool(self.params.get("synchronous", False)) @@ -138,6 +140,26 @@ class Node: """The node's id within its flow.""" return self.name + # ------------------------------------------------------------------------- + # Lifecycle + # + # A plain logic node has nothing to start or stop. The ones that talk to the + # outside — subscriptions, schedules, webhooks — override these, which is + # how the controller can bring a flow up or down without knowing what any + # particular node type is. + # ------------------------------------------------------------------------- + + async def start(self, app: FastAPI | None = None) -> None: + """Begin whatever this node listens to. Called when its flow starts.""" + + async def stop(self, app: FastAPI | None = None) -> None: + """Undo :meth:`start`. Called before a rebuild, and must be idempotent.""" + + def report_health(self, status: str, detail: str | None = None) -> None: + """Say how this node's connection is doing: ok, degraded or down.""" + if self._on_health is not None: + self._on_health(self, status, detail) + def bind(self, pipeline: Pipeline) -> None: """ Bind this node to a pipeline for external triggering. @@ -428,8 +450,9 @@ class HttpNode(Node): description=( "Shared secret for a webhook, appended to its URL: " "/hooks///. Empty leaves the webhook open " - 'to anyone. Use {"$secret": "name"} to read it from the store.' + "to anyone." ), + json_schema_extra={"x-secret": True}, ) def __init__( @@ -568,6 +591,15 @@ class HttpNode(Node): logger.error("Request error in node '%s': %s", self.name, e) raise + async def start(self, app: FastAPI | None = None) -> None: + """A webhook needs a route; a sender reaches out on its own.""" + if self.mode is HttpNode.Mode.TRIGGER and app is not None: + self.register_route(app) + + async def stop(self, app: FastAPI | None = None) -> None: + if self.mode is HttpNode.Mode.TRIGGER and app is not None: + self.unregister_route(app) + def register_route(self, app: FastAPI) -> None: """ Register this node's HTTP endpoint with a FastAPI application. @@ -822,7 +854,9 @@ class MqttNode(Node): broker_host: str = "localhost" broker_port: int = 1883 username: str | None = None - password: str | None = None + password: str | None = Field( + default=None, json_schema_extra={"x-secret": True} + ) client_id: str | None = None qos: int = 0 retain: bool = False @@ -1005,6 +1039,14 @@ class MqttNode(Node): logger.error("MQTT publish error in node '%s': %s", self.name, e) raise + async def start(self, app: FastAPI | None = None) -> None: + """A subscriber listens; a publisher has nothing to start.""" + if self.mode is MqttNode.Mode.SUBSCRIBER: + await self.start_subscription() + + async def stop(self, app: FastAPI | None = None) -> None: + await self.stop_subscription() + async def start_subscription(self) -> None: """ Start the MQTT subscription for trigger mode nodes. @@ -1087,6 +1129,7 @@ class MqttNode(Node): for topic in self._topic_to_ports: await client.subscribe(topic, qos=self.qos) logger.info("[%s] Subscribed to %s", self.name, topic) + self.report_health("ok") async for message in client.messages: if self._stop_event and self._stop_event.is_set(): @@ -1152,6 +1195,7 @@ class MqttNode(Node): self.name, e, ) + self.report_health("down", str(e)) if not (self._stop_event and self._stop_event.is_set()): # Reconnect after a delay await asyncio.sleep(5) @@ -1311,7 +1355,7 @@ class InfluxDbNode(Node): model_config = ConfigDict(extra="allow") url: str - token: str + token: str = Field(json_schema_extra={"x-secret": True}) org: str bucket: str write_precision: str = "ms" @@ -1809,6 +1853,15 @@ class DelayNode(Node): # Cron scheduler # ----------------------------------------------------------------- + async def start(self, app: FastAPI | None = None) -> None: + """Only a node with a schedule has anything to run on its own.""" + if self.cron_expr: + await self.start_cron() + + async def stop(self, app: FastAPI | None = None) -> None: + if self.cron_expr: + await self.stop_cron() + async def start_cron(self) -> None: """ Start the cron scheduler. diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index 3c286b3..ab11c9e 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -258,6 +258,33 @@ class Pipeline: def _timestamp_key(self, msg_name: str) -> str: return f"__ts__:{msg_name}" + def _delivered_key(self, node_name: str, msg_name: str) -> str: + """When a rate-limited input last woke this node.""" + return f"__in_ts__:{node_name}:{msg_name}" + + def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]: + """Drop the outputs whose port is not due to publish yet. + + The value is not lost — the port publishes the current one next time it + is due. Nothing declaring an interval means nothing to look up. + """ + limited = { + name: spec.interval + for name, spec in node.provides.items() + if spec.interval > 0 and name in result + } + if not limited: + return result + + now = time.time() + stamps = self._state.get_multi([self._timestamp_key(name) for name in limited]) + return { + name: value + for name, value in result.items() + if name not in limited + or now - (stamps.get(self._timestamp_key(name)) or 0) >= limited[name] + } + def _increment_message_versions(self, outputs: dict[str, Any]) -> None: for msg_name in outputs: self._state.increment(self._version_key(msg_name)) @@ -298,12 +325,47 @@ class Pipeline: return self._state.compare_and_swap_multi(expected, updates) + def _input_is_due(self, node: Node) -> bool: + """Has any rate-limited input waited out its interval? + + A node runs when *any* of its inputs is due, so a slow port next to a + fast one throttles only itself. Nodes without a limited input never get + here. + """ + limited = { + name: spec.interval + for name, spec in node.requires.items() + if spec.interval > 0 + } + if not limited: + return True + + now = time.time() + keys = [self._delivered_key(node.id, name) for name in limited] + stamps = self._state.get_multi(keys) + due = [ + name + for name, interval in limited.items() + if now - (stamps.get(self._delivered_key(node.id, name)) or 0) >= interval + ] + # An unlimited input alongside a held-back one still wakes the node. + if not due and len(limited) == len(node.requires): + return False + + with self._state.lock(): + for name in due: + self._state[self._delivered_key(node.id, name)] = now + return True + def _is_node_ready(self, node: Node, state: StateBackend) -> bool: with state.lock(): for msg_name in node.requires: if msg_name not in state: return False + if not self._input_is_due(node): + return False + if not node.synchronous: return True @@ -349,6 +411,9 @@ class Pipeline: result = node.execute(inputs) self.publish_log(node, collected, "") + if result: + result = self._throttled(node, result) + if result: ts = time.time() with state.lock(): @@ -510,13 +575,18 @@ class Pipeline: down anyway, and anything still arriving from another thread would be work the flow was explicitly told not to do. A *paused* flow still publishes, so the incoming value is visible on the canvas, and holds - the nodes downstream of it for stepping. + the nodes downstream of it. """ state = self._state if node.flow in self._disabled: return state + if outputs: + # This is where a chatty subscriber gets thinned out, so a port set + # to publish every 60s does so whatever the broker sends. + outputs = self._throttled(node, outputs) + if outputs: ts = time.time() with state.lock(): diff --git a/backend/app/flow/plugins.py b/backend/app/flow/plugins.py new file mode 100644 index 0000000..91e929b --- /dev/null +++ b/backend/app/flow/plugins.py @@ -0,0 +1,77 @@ +"""Loading connector plugins from installed packages. + +A connector is an ordinary Python package that advertises its node class under +the ``fluksio.node_types`` entry point group. The entry point's name is the +node type as it appears on the canvas, and the package's own metadata — name +and version — is the manifest; there is no second format to keep in step. + +Discovery happens once, at startup: a connector's code is imported, and Python +does not re-import a changed module, so a rescan would promise more than it +delivers. Installing or upgrading one means restarting the engine. +""" + +from __future__ import annotations + +import logging +from importlib.metadata import entry_points + +from app.flow.connector import CONTRACT_VERSION, ConnectorNode + +logger = logging.getLogger(__name__) + +ENTRY_POINT_GROUP = "fluksio.node_types" + + +def load_plugins() -> list[str]: + """Register every installed connector. Returns the type names it added.""" + from app.flow.controller import NODE_TYPES, NodeType, _schema_of + + added: list[str] = [] + for entry in entry_points(group=ENTRY_POINT_GROUP): + if entry.name in NODE_TYPES: + logger.warning( + "Ignoring connector '%s': that node type already exists", entry.name + ) + continue + try: + cls = entry.load() + except Exception: + logger.exception("Could not load connector '%s'", entry.name) + continue + + if not (isinstance(cls, type) and issubclass(cls, ConnectorNode)): + logger.warning( + "Ignoring connector '%s': %r is not a ConnectorNode", entry.name, cls + ) + continue + # Checked on the class itself: inheriting a version from a base class + # would let a connector written against an older contract look current. + if cls.__dict__.get("contract") != CONTRACT_VERSION: + logger.warning( + "Ignoring connector '%s': written for contract version %s, " + "this engine speaks %s", + entry.name, + cls.__dict__.get("contract"), + CONTRACT_VERSION, + ) + continue + + NODE_TYPES[entry.name] = NodeType( + title=cls.title or entry.name, + description=cls.description, + cls=cls, + params_schema=_schema_of(cls), + plugin=_provenance(entry), + ) + added.append(entry.name) + logger.info("Loaded connector '%s' from %s", entry.name, _provenance(entry)) + + return added + + +def _provenance(entry: object) -> str: + """Which package this node type came from, for the editor to show.""" + dist = getattr(entry, "dist", None) + if dist is None: + return "unknown" + return f"{dist.name} {dist.version}" diff --git a/backend/app/flow/schemas.py b/backend/app/flow/schemas.py index fd24c83..71b7aec 100644 --- a/backend/app/flow/schemas.py +++ b/backend/app/flow/schemas.py @@ -7,7 +7,7 @@ A flow is structure plus code: this module is the structure. Node logic for from __future__ import annotations import re -from typing import Any +from typing import Any, Literal from pydantic import BaseModel, Field, field_validator @@ -41,6 +41,9 @@ class NodeDef(BaseModel): params: dict[str, Any] = Field(default_factory=dict) requires: list[MessageSpec] = Field(default_factory=list) provides: list[MessageSpec] = Field(default_factory=list) + #: Name of a shared source in the library, instead of this node's own file. + #: Editing it edits the copy every flow using it runs. + source_ref: str | None = None @field_validator("id") @classmethod @@ -76,12 +79,17 @@ class NodeSource(BaseModel): code: str +Health = Literal["ok", "degraded", "down"] + + class NodeStatusPublic(BaseModel): - """Whether a node loaded, and why not.""" + """Whether a node loaded, and how its connection is doing.""" id: str status: str = "active" error: str | None = None + health: Health = "ok" + health_detail: str | None = None class MessageValue(BaseModel): @@ -125,6 +133,13 @@ class FlowsPublic(BaseModel): count: int +class LibraryNode(BaseModel): + """A node source shared across flows, and who is using it.""" + + name: str + used_by: list[str] = Field(default_factory=list) + + class FlowStatePublic(BaseModel): values: dict[str, MessageValue] = Field(default_factory=dict) nodes: list[NodeStatusPublic] = Field(default_factory=list) @@ -138,3 +153,5 @@ class NodeTypeInfo(BaseModel): description: str params_schema: dict[str, Any] = Field(default_factory=dict) has_source: bool = False + #: The package a connector came from; empty for the built-in types. + plugin: str | None = None diff --git a/backend/app/flow/store.py b/backend/app/flow/store.py index 57a319f..97c101f 100644 --- a/backend/app/flow/store.py +++ b/backend/app/flow/store.py @@ -25,6 +25,9 @@ from app.flow.schemas import FlowDef logger = logging.getLogger(__name__) +#: Where the sources of nodes shared between flows live. +LIB_DIR = "_lib" + DEFAULT_SOURCE = '''"""A new node. Return a dict keyed by your output ports.""" @@ -51,6 +54,24 @@ class FlowExists(ValueError): return f"There is already a flow named '{self.name}'" +class LibNotFound(KeyError): + def __init__(self, name: str) -> None: + super().__init__(name) + self.name = name + + def __str__(self) -> str: + return f"No shared node named '{self.name}'" + + +class LibExists(ValueError): + def __init__(self, name: str) -> None: + super().__init__(name) + self.name = name + + def __str__(self) -> str: + return f"There is already a shared node named '{self.name}'" + + class StaleVersion(ValueError): """Someone else saved this flow since the version being written was read.""" @@ -136,6 +157,109 @@ class FlowStore: def _runtime_file(self, name: str) -> Path: return self._flow_dir(name) / "runtime.json" + def _lib_file(self, name: str) -> Path: + return self.root / LIB_DIR / f"{name}.py" + + # ------------------------------------------------------------------------- + # Shared node sources + # + # A node marked reusable moves its code to ``_lib/``; the flows using it + # keep their own ports and parameters and point at it by name, so one fix + # reaches all of them. ``_lib`` cannot collide with a flow — a flow name has + # to start with a letter — and holds no ``flow.json``, so it is invisible to + # everything that lists flows. + # ------------------------------------------------------------------------- + + def list_lib(self) -> list[str]: + return sorted(path.stem for path in (self.root / LIB_DIR).glob("*.py")) + + def lib_exists(self, name: str) -> bool: + return self._lib_file(name).exists() + + def read_lib_source(self, name: str) -> str: + path = self._lib_file(name) + if not path.exists(): + raise LibNotFound(name) + return path.read_text() + + def write_lib_source(self, name: str, code: str) -> bool: + path = self._lib_file(name) + if path.exists() and path.read_text() == code: + return False + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(code) + self._commit(f"Update shared node '{name}'") + return True + + def delete_lib_source(self, name: str) -> None: + path = self._lib_file(name) + if not path.exists(): + raise LibNotFound(name) + path.unlink() + self._commit(f"Delete shared node '{name}'") + + def share_node(self, flow: str, node_id: str, lib_name: str) -> None: + """Move a node's source into the library and point the node at it.""" + with self._write_lock: + if self.lib_exists(lib_name): + raise LibExists(lib_name) + + code = self.read_node_source(flow, node_id, draft=True) + self._lib_file(lib_name).parent.mkdir(parents=True, exist_ok=True) + self._lib_file(lib_name).write_text(code) + + for path in (self._node_file(flow, node_id), + self._draft_node_file(flow, node_id)): + if path.exists(): + path.unlink() + + self._set_source_ref(flow, node_id, lib_name) + self._commit(f"Share node '{flow}.{node_id}' as '{lib_name}'") + + def unshare_node(self, flow: str, node_id: str) -> None: + """Take a private copy of the shared source back into the flow.""" + with self._write_lock: + definition = self.read_flow(flow, draft=True) + node = next((n for n in definition.nodes if n.id == node_id), None) + if node is None or not node.source_ref: + return + code = self.read_lib_source(node.source_ref) + # Written as a draft: the running flow keeps the shared code until + # this is published, like every other edit. + self._draft_node_file(flow, node_id).parent.mkdir( + parents=True, exist_ok=True + ) + self._draft_node_file(flow, node_id).write_text(code) + self._set_source_ref(flow, node_id, None) + self._commit(f"Unshare node '{flow}.{node_id}'") + + def usages(self, lib_name: str) -> list[str]: + """Every ``flow.node`` pointing at this shared source.""" + used = [] + for name in self.list_flows(): + try: + definition = self.read_flow(name, draft=True) + except FlowNotFound: + continue + used += [ + f"{name}.{node.id}" + for node in definition.nodes + if node.source_ref == lib_name + ] + return sorted(used) + + def _set_source_ref(self, flow: str, node_id: str, ref: str | None) -> None: + """Point a node at a shared source, in the draft the editor is on.""" + definition = self.read_flow(flow, draft=True) + definition.nodes = [ + node.model_copy(update={"source_ref": ref}) if node.id == node_id else node + for node in definition.nodes + ] + stored = definition.model_copy(update={"version": definition.version + 1}) + path = self._draft_file(flow) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(stored.model_dump_json(indent=2) + "\n") + # ------------------------------------------------------------------------- # Runtime state # ------------------------------------------------------------------------- diff --git a/backend/app/main.py b/backend/app/main.py index 52fd0ef..748e28e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,7 @@ from app.core.config import settings from app.flow import logs from app.flow.controller import FlowController from app.flow.events import event_bus +from app.flow.plugins import load_plugins from app.flow.secrets import init_secrets from app.flow.state import MemoryState, RedisState, StateBackend from app.flow.store import FlowStore @@ -38,6 +39,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # Node code is user code, and `print` is how it says things. logs.install() init_secrets(settings.SECRETS_FILE, settings.SECRET_KEY) + # Connectors register their node types before any flow is built with them. + load_plugins() controller = FlowController( store=FlowStore(settings.FLOWS_DIR), diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py new file mode 100644 index 0000000..18c4691 --- /dev/null +++ b/backend/tests/flow/test_connector.py @@ -0,0 +1,156 @@ +"""The connector contract: polling, deduplication, health and discovery.""" + +import asyncio +from typing import Any + +from app.flow.connector import CONTRACT_VERSION, ConnectorNode +from app.flow.controller import NODE_TYPES +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import Node +from app.flow.pipeline import Pipeline +from app.flow.plugins import load_plugins + + +class Sensor(ConnectorNode): + contract = CONTRACT_VERSION + title = "Test sensor" + description = "Reads whatever it is told to." + + class Params(ConnectorNode.Params): + secret_token: str | None = None + + def __init__(self, readings: list[Any] | None = None, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._readings = list(readings or []) + self.polls = 0 + + async def poll(self) -> dict[str, Any] | None: + self.polls += 1 + if not self._readings: + return None + value = self._readings.pop(0) + if isinstance(value, Exception): + raise value + return {"reading": value} + + +def a_sensor(readings: list[Any], **params: Any) -> Sensor: + node = Sensor( + readings=readings, + provides=[MessageSpec(name="reading", dtype=DType.FLOAT)], + params={"poll_interval": 0.01, **params}, + ) + node.assign_flow("demo", "sensor") + return node + + +def run_briefly(node: ConnectorNode, seconds: float = 0.12) -> None: + """Start the poll loop, let it tick a few times, stop it.""" + + async def cycle() -> None: + await node.start() + await asyncio.sleep(seconds) + await node.stop() + + asyncio.run(cycle()) + + +def test_polling_publishes_what_it_reads(): + node = a_sensor([21.5]) + pipeline = Pipeline(nodes=[node]) + + run_briefly(node) + + assert pipeline.state["demo.reading"] == 21.5 + + +def test_an_unchanged_reading_is_not_republished(): + node = a_sensor([21.5, 21.5, 21.5]) + consumer_ran: list[float] = [] + + def consume(reading, params): + consumer_ran.append(reading) + return None + + consumer = Node( + f=consume, + requires=[MessageSpec(name="reading", dtype=DType.FLOAT)], + name="consumer", + ) + consumer.assign_flow("demo", "consumer") + Pipeline(nodes=[node, consumer]) + + run_briefly(node) + + # Polled repeatedly, but the value never changed, so downstream ran once. + assert node.polls > 1 + assert consumer_ran == [21.5] + + +def test_a_failing_poll_reports_down_and_keeps_going(): + health: list[tuple[str, str | None]] = [] + node = a_sensor([RuntimeError("device unplugged"), 21.5]) + node._on_health = lambda _node, status, detail: health.append((status, detail)) + Pipeline(nodes=[node]) + + run_briefly(node) + + assert ("down", "RuntimeError: device unplugged") in health + # It recovered rather than giving up. + assert health[-1][0] == "ok" + + +def test_a_credential_param_is_marked_for_the_editor(): + schema = Sensor.Params.model_json_schema() + assert schema["properties"]["poll_interval"]["default"] == 0 + assert "secret_token" in schema["properties"] + + +def test_a_connector_is_discovered_from_its_entry_point(monkeypatch): + class FakeDist: + name = "fluksio-connector-test" + version = "0.1.0" + + class FakeEntry: + name = "test_sensor" + dist = FakeDist() + + def load(self): + return Sensor + + monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()]) + try: + assert load_plugins() == ["test_sensor"] + assert NODE_TYPES["test_sensor"].plugin == "fluksio-connector-test 0.1.0" + assert NODE_TYPES["test_sensor"].title == "Test sensor" + finally: + NODE_TYPES.pop("test_sensor", None) + + +def test_a_connector_written_for_another_contract_is_refused(monkeypatch): + class Outdated(ConnectorNode): + contract = CONTRACT_VERSION + 1 + + class FakeEntry: + name = "outdated" + dist = None + + def load(self): + return Outdated + + monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()]) + assert load_plugins() == [] + assert "outdated" not in NODE_TYPES + + +def test_a_connector_may_not_take_over_a_built_in_type(monkeypatch): + class FakeEntry: + name = "mqtt" + dist = None + + def load(self): # pragma: no cover - never reached + raise AssertionError("should not be loaded") + + monkeypatch.setattr("app.flow.plugins.entry_points", lambda group: [FakeEntry()]) + assert load_plugins() == [] + assert NODE_TYPES["mqtt"].plugin is None diff --git a/backend/tests/flow/test_discretization.py b/backend/tests/flow/test_discretization.py new file mode 100644 index 0000000..86e98c6 --- /dev/null +++ b/backend/tests/flow/test_discretization.py @@ -0,0 +1,100 @@ +"""Per-port intervals: deliver at most every x seconds.""" + +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import Node +from app.flow.pipeline import Pipeline + + +def spec(name: str, interval: float = 0) -> MessageSpec: + return MessageSpec(name=name, dtype=DType.FLOAT, interval=interval) + + +def make_node(node_id: str, f, requires=(), provides=()) -> Node: + node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id) + node.assign_flow("demo", node_id) + return node + + +def test_a_limited_output_publishes_once_inside_its_window(): + readings = iter([1.0, 2.0, 3.0]) + source = make_node( + "source", + lambda params: {"temp": next(readings)}, + provides=[spec("temp", interval=60)], + ) + pipeline = Pipeline(nodes=[source]) + + pipeline.run({}) + assert pipeline.state["demo.temp"] == 1.0 + + # Same window: the reading is taken but not published. + pipeline.run({}) + assert pipeline.state["demo.temp"] == 1.0 + + +def test_an_unlimited_output_publishes_every_time(): + readings = iter([1.0, 2.0]) + source = make_node( + "source", + lambda params: {"temp": next(readings)}, + provides=[spec("temp")], + ) + pipeline = Pipeline(nodes=[source]) + + pipeline.run({}) + pipeline.run({}) + assert pipeline.state["demo.temp"] == 2.0 + + +def test_a_limited_input_wakes_its_node_once_inside_the_window(): + seen: list[float] = [] + source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")]) + consumer = make_node( + "consumer", + lambda temp, params: seen.append(temp), + requires=[spec("temp", interval=60)], + ) + # Binding the nodes is what the pipeline is for here. + Pipeline(nodes=[source, consumer]) + + source.inject({"temp": 20.0}) + source.inject({"temp": 21.0}) + + assert seen == [20.0] + + +def test_an_unthrottled_input_still_wakes_a_node_beside_a_throttled_one(): + seen: list[tuple[float, float]] = [] + fast = make_node("fast", lambda params: None, provides=[spec("quick")]) + slow = make_node("slow", lambda params: None, provides=[spec("rare")]) + consumer = make_node( + "consumer", + lambda quick, rare, params: seen.append((quick, rare)), + requires=[spec("quick"), spec("rare", interval=60)], + ) + pipeline = Pipeline(nodes=[fast, slow, consumer]) + pipeline.state["demo.rare"] = 1.0 + + fast.inject({"quick": 1.0}) + fast.inject({"quick": 2.0}) + + # The throttled port holds back only itself. + assert [quick for quick, _ in seen] == [1.0, 2.0] + + +def test_a_manual_run_is_never_throttled_on_its_inputs(): + seen: list[float] = [] + source = make_node("source", lambda params: {"temp": 20.0}, provides=[spec("temp")]) + consumer = make_node( + "consumer", + lambda temp, params: seen.append(temp), + requires=[spec("temp", interval=3600)], + ) + pipeline = Pipeline(nodes=[source, consumer]) + + # Pressing Run is an explicit ask; the interval governs the flow's own + # traffic, not what the person in front of it asked for. + pipeline.run({}) + pipeline.run({}) + + assert len(seen) == 2 diff --git a/backend/tests/flow/test_library.py b/backend/tests/flow/test_library.py new file mode 100644 index 0000000..72bb336 --- /dev/null +++ b/backend/tests/flow/test_library.py @@ -0,0 +1,96 @@ +"""Nodes shared across flows: one source, many instances.""" + +from pathlib import Path + +import pytest + +from app.flow.messages import MessageSpec +from app.flow.schemas import FlowDef, NodeDef +from app.flow.store import FlowStore, LibExists, LibNotFound + +SOURCE = "def process(params):\n return {'temp': 1}\n" + + +@pytest.fixture +def store(tmp_path: Path) -> FlowStore: + return FlowStore(tmp_path / "flows") + + +def a_flow(name: str = "heating") -> FlowDef: + return FlowDef( + name=name, + nodes=[NodeDef(id="sensor", provides=[MessageSpec(name="temp")])], + ) + + +def test_sharing_moves_the_source_and_points_the_node_at_it(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + + store.share_node("heating", "sensor", "read_temp") + + assert store.list_lib() == ["read_temp"] + assert store.read_lib_source("read_temp") == SOURCE + node = store.read_flow("heating", draft=True).nodes[0] + assert node.source_ref == "read_temp" + # The private copy is gone; the library one is what it runs. + assert not (store.root / "heating" / "nodes" / "sensor.py").exists() + + +def test_the_library_is_not_mistaken_for_a_flow(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + store.share_node("heating", "sensor", "read_temp") + + assert store.list_flows() == ["heating"] + assert [flow.name for flow in store.read_all()] == ["heating"] + + +def test_a_second_flow_can_use_the_same_source(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + store.share_node("heating", "sensor", "read_temp") + + store.write_flow( + FlowDef( + name="cooling", + nodes=[ + NodeDef( + id="sensor", + source_ref="read_temp", + provides=[MessageSpec(name="temp")], + ) + ], + ) + ) + + assert store.usages("read_temp") == ["cooling.sensor", "heating.sensor"] + + +def test_a_shared_name_is_not_taken_twice(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + store.share_node("heating", "sensor", "read_temp") + + store.write_flow(FlowDef(name="cooling", nodes=[NodeDef(id="sensor")])) + with pytest.raises(LibExists): + store.share_node("cooling", "sensor", "read_temp") + + +def test_unsharing_takes_a_private_copy_back(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + store.share_node("heating", "sensor", "read_temp") + + store.unshare_node("heating", "sensor") + + node = store.read_flow("heating", draft=True).nodes[0] + assert node.source_ref is None + assert store.read_node_source("heating", "sensor", draft=True) == SOURCE + # The library keeps its copy for whoever else is using it. + assert store.list_lib() == ["read_temp"] + + +def test_a_missing_shared_source_is_reported(store: FlowStore): + with pytest.raises(LibNotFound): + store.read_lib_source("nothing_here") diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 9aa50cc..d1f9d36 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -343,6 +343,26 @@ export const HistoryPointSchema = { description: 'One numeric value a message carried, and when.' } as const; +export const LibraryNodeSchema = { + properties: { + name: { + type: 'string', + title: 'Name' + }, + used_by: { + items: { + type: 'string' + }, + type: 'array', + title: 'Used By' + } + }, + type: 'object', + required: ['name'], + title: 'LibraryNode', + description: 'A node source shared across flows, and who is using it.' +} as const; + export const MessageSchema = { properties: { message: { @@ -398,6 +418,12 @@ export const MessageSpecSchema = { dtype: { '$ref': '#/components/schemas/DType', default: 'float' + }, + interval: { + type: 'number', + minimum: 0, + title: 'Interval', + default: 0 } }, type: 'object', @@ -408,7 +434,11 @@ export const MessageSpecSchema = { the flow name at load time; empty means the port is unbound. :param port: The identifier the node function sees. Defaults to the last segment of \`\`name\`\`, so unqualified flows read naturally. -:param dtype: Payload type, validated on every message that passes through.` +:param dtype: Payload type, validated on every message that passes through. +:param interval: Deliver at most every this many seconds; 0 is every time. + On an output it holds back publishing, on an input it holds back waking + the node. The value is never lost — state keeps the latest — only the + delivery is skipped.` } as const; export const MessageValueSchema = { @@ -492,6 +522,17 @@ export const NodeDef_InputSchema = { }, type: 'array', title: 'Provides' + }, + source_ref: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Source Ref' } }, type: 'object', @@ -541,6 +582,17 @@ export const NodeDef_OutputSchema = { }, type: 'array', title: 'Provides' + }, + source_ref: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Source Ref' } }, type: 'object', @@ -583,12 +635,29 @@ export const NodeStatusPublicSchema = { } ], title: 'Error' + }, + health: { + type: 'string', + enum: ['ok', 'degraded', 'down'], + title: 'Health', + default: 'ok' + }, + health_detail: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Health Detail' } }, type: 'object', required: ['id'], title: 'NodeStatusPublic', - description: 'Whether a node loaded, and why not.' + description: 'Whether a node loaded, and how its connection is doing.' } as const; export const NodeTypeInfoSchema = { @@ -614,6 +683,17 @@ export const NodeTypeInfoSchema = { type: 'boolean', title: 'Has Source', default: false + }, + plugin: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Plugin' } }, type: 'object', @@ -733,6 +813,18 @@ export const SecretValueSchema = { title: 'SecretValue' } as const; +export const ShareRequestSchema = { + properties: { + lib_name: { + type: 'string', + title: 'Lib Name' + } + }, + type: 'object', + required: ['lib_name'], + title: 'ShareRequest' +} as const; + export const TokenSchema = { properties: { access_token: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 25bd75d..4ddbf35 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen'; +import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse } from './types.gen'; export class FlowsService { /** @@ -32,6 +32,40 @@ export class FlowsService { }); } + /** + * Read Library + * The node sources shared across flows, and which nodes use each. + * @returns LibraryNode Successful Response + * @throws ApiError + */ + public static readLibrary(): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/flows/library' + }); + } + + /** + * Delete Shared Node + * Remove a shared source, as long as no flow still runs it. + * @param data The data for the request. + * @param data.libName + * @returns Message Successful Response + * @throws ApiError + */ + public static deleteSharedNode(data: FlowsDeleteSharedNodeData): CancelablePromise { + return __request(OpenAPI, { + method: 'DELETE', + url: '/api/v1/flows/library/{lib_name}', + path: { + lib_name: data.libName + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Flow * Read one flow, with the state of its nodes. @@ -202,6 +236,10 @@ export class FlowsService { * The answer comes from compiling the code rather than from the running * pipeline: a draft is not deployed, and compiling is both faster and more * precise about what the author just typed. + * + * A shared node writes to the library, so the fix reaches every flow using + * it — and that one is live immediately rather than waiting for a publish, + * because the copy is not any single flow's to hold back. * @param data The data for the request. * @param data.name * @param data.nodeId @@ -225,6 +263,55 @@ export class FlowsService { }); } + /** + * Share Node + * Move this node's code into the library so other flows can run it too. + * @param data The data for the request. + * @param data.name + * @param data.nodeId + * @param data.requestBody + * @returns FlowDetail Successful Response + * @throws ApiError + */ + public static shareNode(data: FlowsShareNodeData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/nodes/{node_id}/share', + path: { + name: data.name, + node_id: data.nodeId + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Unshare Node + * Take a private copy of the shared code back into this flow. + * @param data The data for the request. + * @param data.name + * @param data.nodeId + * @returns FlowDetail Successful Response + * @throws ApiError + */ + public static unshareNode(data: FlowsUnshareNodeData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/nodes/{node_id}/unshare', + path: { + name: data.name, + node_id: data.nodeId + }, + errors: { + 422: 'Validation Error' + } + }); + } + /** * Start Flow * Let the engine run this flow again. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 0178eec..4886c81 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -105,6 +105,14 @@ export type HTTPValidationError = { detail?: Array; }; +/** + * A node source shared across flows, and who is using it. + */ +export type LibraryNode = { + name: string; + used_by?: Array<(string)>; +}; + export type Message = { message: string; }; @@ -129,11 +137,16 @@ export type MessageHistory = { * :param port: The identifier the node function sees. Defaults to the last * segment of ``name``, so unqualified flows read naturally. * :param dtype: Payload type, validated on every message that passes through. + * :param interval: Deliver at most every this many seconds; 0 is every time. + * On an output it holds back publishing, on an input it holds back waking + * the node. The value is never lost — state keeps the latest — only the + * delivery is skipped. */ export type MessageSpec = { name?: string; port?: string; dtype?: DType; + interval?: number; }; /** @@ -162,6 +175,7 @@ export type NodeDef_Input = { }; requires?: Array; provides?: Array; + source_ref?: (string | null); }; /** @@ -177,6 +191,7 @@ export type NodeDef_Output = { }; requires?: Array; provides?: Array; + source_ref?: (string | null); }; /** @@ -187,14 +202,18 @@ export type NodeSource = { }; /** - * Whether a node loaded, and why not. + * Whether a node loaded, and how its connection is doing. */ export type NodeStatusPublic = { id: string; status?: string; error?: (string | null); + health?: 'ok' | 'degraded' | 'down'; + health_detail?: (string | null); }; +export type health = 'ok' | 'degraded' | 'down'; + /** * A node type the editor can offer, with its parameter schema. */ @@ -206,6 +225,7 @@ export type NodeTypeInfo = { [key: string]: unknown; }; has_source?: boolean; + plugin?: (string | null); }; /** @@ -246,6 +266,10 @@ export type SecretValue = { value: string; }; +export type ShareRequest = { + lib_name: string; +}; + export type Token = { access_token: string; token_type?: string; @@ -332,6 +356,14 @@ export type FlowsReadFlowsResponse = (FlowsPublic); export type FlowsReadNodeTypesResponse = (Array); +export type FlowsReadLibraryResponse = (Array); + +export type FlowsDeleteSharedNodeData = { + libName: string; +}; + +export type FlowsDeleteSharedNodeResponse = (Message); + export type FlowsReadFlowData = { name: string; }; @@ -386,6 +418,21 @@ export type FlowsSaveNodeSourceData = { export type FlowsSaveNodeSourceResponse = (NodeStatusPublic); +export type FlowsShareNodeData = { + name: string; + nodeId: string; + requestBody: ShareRequest; +}; + +export type FlowsShareNodeResponse = (FlowDetail); + +export type FlowsUnshareNodeData = { + name: string; + nodeId: string; +}; + +export type FlowsUnshareNodeResponse = (FlowDetail); + export type FlowsStartFlowData = { name: string; }; diff --git a/frontend/src/components/Flow/CommandPalette.tsx b/frontend/src/components/Flow/CommandPalette.tsx index ac18b26..91f3edf 100644 --- a/frontend/src/components/Flow/CommandPalette.tsx +++ b/frontend/src/components/Flow/CommandPalette.tsx @@ -1,3 +1,4 @@ +import { useQuery } from "@tanstack/react-query" import { useNavigate } from "@tanstack/react-router" import { useEffect } from "react" @@ -10,6 +11,7 @@ import { CommandItem, CommandList, } from "@/components/ui/command" +import { libraryQueryOptions } from "./queries" /** * ⌘K: add a node, jump to another flow, or run the current one, without @@ -21,6 +23,7 @@ export function CommandPalette({ nodeTypes, flows, onAddNode, + onAddSharedNode, onRun, }: { open: boolean @@ -28,9 +31,14 @@ export function CommandPalette({ nodeTypes: NodeTypeInfo[] flows: FlowSummary[] onAddNode: (type: string) => void + onAddSharedNode: (libName: string) => void onRun: () => void }) { const navigate = useNavigate() + const { data: libraryData } = useQuery({ + ...libraryQueryOptions(), + enabled: open, + }) useEffect(() => { const onKey = (event: KeyboardEvent) => { @@ -48,6 +56,8 @@ export function CommandPalette({ // dialog outright is deterministic; the palette does not need to fade out. if (!open) return null + const library = libraryData ?? [] + const close = (action: () => void) => { onOpenChange(false) action() @@ -81,6 +91,28 @@ export function CommandPalette({ ))} + {library.length > 0 ? ( + + {library.map((entry) => ( + close(() => onAddSharedNode(entry.name))} + > + + {entry.name} + + {(entry.used_by ?? []).length === 1 + ? "Used once" + : `Used ${(entry.used_by ?? []).length} times`} + , code shared with every flow using it + + + + ))} + + ) : null} + {flows.length > 0 ? ( {flows.map((flow) => ( diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index c180b0d..ff38cd2 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -453,8 +453,8 @@ function FlowEditorInner({ }) const addNode = useCallback( - (type: string) => { - const id = uniqueNodeId(definitions, type) + (type: string, sourceRef?: string) => { + const id = uniqueNodeId(definitions, sourceRef ?? type) // Drop it where the user is looking, but never on top of another node. const position = freePosition( definitions, @@ -470,6 +470,9 @@ function FlowEditorInner({ params: {}, requires: [], provides: [], + // A shared node brings the code; the ports and settings are this + // flow's own. + ...(sourceRef ? { source_ref: sourceRef } : {}), } const nextDefinitions = [...definitions, node] const nextCanvas = [ @@ -809,6 +812,9 @@ function FlowEditorInner({ onSaveSource={(code) => { if (selected) sourceMutation.mutate({ nodeId: selected.id, code }) }} + // Sharing rewrites the stored document, so the canvas takes the + // server's copy rather than keeping its own. + onShared={onReload} onClose={() => { flush() setEditorExpanded(false) @@ -830,6 +836,7 @@ function FlowEditorInner({ nodeTypes={nodeTypeInfo ?? []} flows={flows.data} onAddNode={addNode} + onAddSharedNode={(libName) => addNode("python", libName)} onRun={() => runMutation.mutate()} /> diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index 0707eee..3707469 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -1,8 +1,14 @@ -import { useQuery } from "@tanstack/react-query" +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" import { Maximize2, Minimize2, X } from "lucide-react" import { lazy, Suspense, useEffect, useRef, useState } from "react" -import type { DType, MessageSpec, NodeDef_Input, NodeTypeInfo } from "@/client" +import { + type DType, + FlowsService, + type MessageSpec, + type NodeDef_Input, + type NodeTypeInfo, +} from "@/client" import { Button } from "@/components/ui/button" import { Command, @@ -22,15 +28,24 @@ import { SelectValue, } from "@/components/ui/select" import { Switch } from "@/components/ui/switch" +import useCustomToast from "@/hooks/useCustomToast" import { cn } from "@/lib/utils" import { MessageSparkline } from "./MessageSparkline" -import { nodeSourceQueryOptions } from "./queries" +import { + flowKeys, + libraryQueryOptions, + nodeSourceQueryOptions, + secretsQueryOptions, +} from "./queries" import { PANEL_SECTION, SidePanel } from "./SidePanel" const NodeEditor = lazy(() => import("./NodeEditor")) const DTYPES: DType[] = ["float", "int", "str", "bool", "json"] +/** Radix selects cannot hold an empty value, so "no secret" needs a name. */ +const NO_SECRET = "__none__" + const SECTION = PANEL_SECTION /** @@ -208,6 +223,19 @@ function PortList({ ))} + + update(index, { interval: Number(event.target.value) || 0 }) + } + /> + + ) + } + + return ( +
+ Reuse +

+ Move this node's code to the library so other flows can run it too. +

+
+ setName(event.target.value)} + /> + +
+
+ ) +} + function PanelBody({ node, flow, @@ -316,6 +484,7 @@ function PanelBody({ onChange, onRenameMessage, onSaveSource, + onShared, onToggleExpand, }: { node: NodeDef_Input @@ -326,6 +495,7 @@ function PanelBody({ onChange: (next: NodeDef_Input) => void onRenameMessage: (previous: string, next: string) => void onSaveSource: (code: string) => void + onShared: () => void onToggleExpand: () => void }) { const hasSource = nodeType?.has_source ?? node.type === "python" @@ -387,12 +557,17 @@ function PanelBody({ params={node.params ?? {}} onChange={(params) => onChange({ ...node, params })} /> + {hasSource ? ( + + ) : null} {hasSource ? (
- Code + + {node.source_ref ? `Shared code · ${node.source_ref}` : "Code"} +