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 <noreply@anthropic.com>

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
 # -----------------------------------------------------------------------------

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)

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]
 

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:

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/<flow>/<url>/<secret>. 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.

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():

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}"

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

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
     # -------------------------------------------------------------------------

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),