Back to App

Workflow Engine

The server-side workflow engine (workflow_engine.py) executes node editor workflows independently of the browser. Workflows continue running even after the user closes their tab.

Execution Modes

Mode Description
oneshot Run all nodes once, top-to-bottom (topological sort)
scheduled Cron-style recurring execution (30s, 5m, 1h, 24h)
live (planned) MQTT source nodes push data continuously

How It Works

  1. User clicks Deploy (🚀) in the node editor
  2. Frontend POSTs the workflow JSON to /api/workflow/deploy
  3. WorkflowManager creates a WorkflowExecutor thread
  4. Executor performs topological sort on the node graph
  5. Nodes execute in order, passing data through wired connections
  6. Each node's execution is logged to workflow_logs table
  7. Run status is tracked in workflow_runs table
  8. SSE events broadcast real-time status to the browser

Node Executors (66+)

Core Executors

Category Node Types
Input text-input, text, number-input
AI seti-chat, vllm-chat, ollama-chat, ollama, bill-analyst
Tools web-search, search, http-request, api-request, web-api
Transform json-parser, json-parse, json-path, template, string-template, merge, math
Logic threshold, condition, if-else, delay, timer, loop
Storage neo4j-query, neo4j-store, vault-read, vault-write
Output display, log, debug, notify, alert, email, obsidian-log
Data victron-data, fleet-status

Energy Executors (v9.2.0)

All 14 Energy nodes are registered via _exec_energy_generic() factory that calls /api/energy/<type>:

solar-inverter, battery-monitor, grid-meter, powerwall, energy-forecast, ev-charger, ups-monitor, utility-rates, energy-cost, co2-tracker, load-controller, water-heater, demand-response, powerline-quality

Database Tables

workflow_runs

Column Type Description
id SERIAL Primary key
name TEXT Workflow name
user_id INTEGER Owner (multi-tenancy)
status TEXT running, completed, failed, stopped
mode TEXT oneshot, scheduled
schedule TEXT Schedule interval (e.g., "5m")
node_states JSONB Per-node execution state
started_at TIMESTAMP Run start time
finished_at TIMESTAMP Run end time

workflow_logs

Column Type Description
id SERIAL Primary key
run_id INTEGER FK to workflow_runs
node_id TEXT Node identifier
level TEXT info, ok, warn, error
message TEXT Log message
created_at TIMESTAMP Timestamp

Multi-Tenancy

  • Workflow runs are scoped by user_id
  • Admin sees all runs; regular users see only their own
  • WorkflowExecutor tracks user_id through execution
  • DB index on workflow_runs(user_id) for performance

Training Corpus

Every workflow execution auto-writes a structured markdown summary to Memory/training/wf-{id}.md in Obsidian. This feeds the Seti-Mini SFT training pipeline.

Scheduled Workflows

POST /api/workflow/deploy
{
  "name": "solar-monitor",
  "mode": "scheduled",
  "schedule": "5m"
}

The workflow will execute every 5 minutes until stopped:

POST /api/workflow/stop
{ "name": "solar-monitor" }

SSE Broadcasting

Workflow events are broadcast via Server-Sent Events:

const es = new EventSource('/api/events');
es.addEventListener('workflow', (e) => {
  const data = JSON.parse(e.data);
  // { run_id, node_id, status, message }
});