Add websocket node: client or served socket, listener or sender

ws:// and wss:// URLs dial out and hold a supervised connection; a path is
served under /hooks/<flow>/ with an optional secret. Outputs only listen,
inputs send. Frames are JSON keyed by port, with json_key and on_connect.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011CT3mowxKeZywDCH4fS1gt

backend/fluksio/flow/controller.py

@@ -26,7 +26,7 @@ from collections.abc import Callable
 from contextlib import suppress
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, cast
+from typing import Any, TypeGuard, cast
 
 from fastapi import FastAPI
 from fastapi.concurrency import run_in_threadpool
@@ -54,6 +54,7 @@ from fluksio.flow.nodes import (
     RbeNode,
     SwitchNode,
     TriggerNode,
+    WebSocketNode,
     WrenNode,
 )
 from fluksio.flow.pipeline import (
@@ -93,6 +94,14 @@ logger = logging.getLogger(__name__)
 
 HOOK_PREFIX = "/hooks"
 
+
+def _is_hook(node: Node | None) -> TypeGuard[HttpNode | WebSocketNode]:
+    """Mounted under HOOK_PREFIX: a webhook, or a socket the engine serves."""
+    return (isinstance(node, HttpNode) and node.mode is HttpNode.Mode.TRIGGER) or (
+        isinstance(node, WebSocketNode) and node.serves
+    )
+
+
 # How long a node gets to close what it opened before the rebuild moves on.
 # A node's `stop` talks to whatever it connected to, and a broker that has gone
 # away can leave it waiting for an acknowledgement that never arrives — which
@@ -339,6 +348,12 @@ NODE_TYPES: dict[str, NodeType] = {
         cls=HttpNode,
         params_schema=_schema_of(HttpNode),
     ),
+    "websocket": NodeType(
+        title="WebSocket",
+        description="Listen on a socket, or send to it — as a client or a server.",
+        cls=WebSocketNode,
+        params_schema=_schema_of(WebSocketNode),
+    ),
     "influxdb": NodeType(
         title="InfluxDB",
         description="Write measurements to a bucket, or read them back.",
@@ -1229,8 +1244,8 @@ 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.
+            if _is_hook(node):
+                # Hooks live under their flow, away from the JSON API.
                 node.url = f"{HOOK_PREFIX}/{flow}/{node.url.lstrip('/')}"
 
             entry.node = node
@@ -1876,23 +1891,21 @@ def _collect_issues(
         for entry in loaded.values()
         if entry.status is NodeStatus.ERROR
     ]
-    # Webhooks are mounted without authentication, so a hook without a secret is
+    # Hooks are mounted without authentication, so a hook without a secret is
     # open to anyone who guesses its URL. Flows written before the parameter
     # existed keep running, but say so.
     issues += [
         ValidationIssue(
             code="unauthenticated_hook",
             message=(
-                f"Webhook '{entry.node.local_id}' has no secret — "
-                "anyone who knows its URL can trigger it."
+                f"Hook '{entry.node.local_id}' has no secret — "
+                "anyone who knows its URL can reach it."
             ),
             flow=entry.flow,
             node=entry.id,
         )
         for entry in loaded.values()
-        if isinstance(entry.node, HttpNode)
-        and entry.node.mode == HttpNode.Mode.TRIGGER
-        and not entry.node.secret
+        if _is_hook(entry.node) and not entry.node.secret
     ]
     # A node whose body was never stored runs the new-node template, which
     # returns nothing — quietly, and looking healthy the whole time.