Native Visual Stream Studio

This document explains the Native Visual Stream Studio embedded in rekuiper.

The studio gives users an interactive graphical interface to design, wire, test, and deploy streaming Directed Acyclic Graph (DAG) rules directly from the kuiperd single binary.

Overview

 +-------------------------------------------------------------------------------+
 |                              STREAM DAG STUDIO                                |
 +-------------------+---------------------------------------+-------------------+
 | COMPONENT PALETTE |         INTERACTIVE DAG CANVAS        | INSPECTION DRAWER |
 |                   |                                       |                   |
 | [ Sources ]       |   [MQTT Source]                       | Node Properties:  |
 |  - MQTT Source    |         | (Port 0)                    |  ID: switch_1     |
 |  - Modbus PLC     |         v                             |                   |
 |  - OPC-UA         |   [Switch Router]                     | Wire-Tap Probe:   |
 |                   |    /           \                      |  Status: ACTIVE   |
 | [ Operators ]     |  (P0)          (P1)                   |  Limit: 20 msg/s  |
 |  - Switch Router  |   v             v                     |  Dropped: 0       |
 |  - Rate Limiter   | [Alert MQTT]   [Log Sink]             |                   |
 |  - Watchdog       |                                       | Live Tuple Feed:  |
 |  - Filter / SQL   |                                       | {"temp": 82.4}    |
 |                   |                                       |                   |
 | [ Sinks ]         |                                       | GraphRule JSON:   |
 |  - MQTT Sink      |                                       | Two-way live sync |
 |  - REST / Log     |                                       |                   |
 +-------------------+---------------------------------------+-------------------+

Visual Architecture Diagram

Core Features

1. Interactive DAG Canvas

   • Connect sources, operators, and sinks with mouse interactions.
   • Support for multi-port routing nodes (switch operator with port 0, port 1, and default ports).
   • Real-time cubic Bezier curve rendering with auto-layout alignment.
   • Animated stream pulses across active connecting wires when telemetry tuples flow.

2. Two-Way GraphRule JSON Synchronization

   • Canvas modifications immediately update the serialized GraphRule JSON definition.
   • Text edits in the JSON editor update the visual DAG canvas upon clicking Apply to Canvas.
   • Direct one-click deployment to the running Kuiper engine via the Deploy Pipeline button.

3. Live Node Metric Badges and /comms Wire-Tap

   • Real-time WebSocket connection to /comms.
   • Node badges display total processed records.
   • Click-to-tap wire-tap inspection with a token-bucket rate limiter (20 messages/second).
   • Real-time ⚡ TAP status badge displays when a probe is active.
   • Monospace telemetry inspection terminal for live record inspection.

4. Natural-Language-to-Graph AI Assistant

   • Describe stream flows in plain language (for example: "Read factory MQTT -> rate limit 10/s -> route temp > 75 to alerts_mqtt, else log").
   • Click Synthesize Pipeline to compile the prompt into nodes, properties, and edges.

Real-Time Wire-Tap Telemetry Format

When you enable a wire tap on a node, the daemon streams JSON-RPC events over /comms with the following schema:

{
  "type": "tuple_sample",
  "rule_id": "rule_turbine_monitor",
  "node_id": "switch_1",
  "data": {
    "device_id": "turbine-alpha-01",
    "temperature": 78.4,
    "vibration": 0.042,
    "ts": 1740001250000
  },
  "dropped_since_last": 0,
  "timestamp": 1740001250100
}

• rule_id: Identifier of the running pipeline.
• node_id: Target node identifier with the active probe.
• data: Exact tuple map emitted by the node execution stage.
• dropped_since_last: Number of sample tuples discarded by the token-bucket rate limiter to preserve network bandwidth.

Endpoints

• GET /ui: Serves the embedded single-binary console.
• GET /api/v1/palette: Returns dynamic node definitions and property schemas.
• GET /comms: WebSocket bus for live tuple samples and connector status events.
• POST /rules/:id/tap/:node_id/enable: Activates a token-bucket wire-tap probe on the specified node.
• POST /rules/:id/tap/:node_id/disable: Deactivates the wire-tap probe on the specified node.

Model Context Protocol (MCP) Integration

AI agents integrate with Visual Stream Studio through the following rekuiper-mcp tools:

MCP Tool	Function
get_studio_palette	Retrieves the catalog of visual components, ports, and configuration schemas.
validate_graph_rule	Validates a Visual DAG rule JSON definition offline before deployment.
toggle_wire_tap	Enables or disables a wire-tap probe on any DAG node with configurable rate limiting.
get_wire_tap_status	Queries active wire-tap probe status, message counters, and drop counts.

You can also read the live palette resource via rekuiper://studio/palette or invoke the prompt design_visual_dag_pipeline.