feat(flow): AI node — in-flow LLM completion (Claude/OpenAI/Kimi)

CHANGELOG.md

@@ -5,6 +5,20 @@ versioning: [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- **AI node** — a new flow node that calls an LLM (Claude / OpenAI / Kimi)
+  from inside a DAG. Provider, API keys, and default models resolve from the
+  project secret vault exactly like the in-app assistant
+  (`CLAUDE_API_KEY` / `OPENAI_API_KEY` / `KIMI_API_KEY`,
+  `*_DEFAULT_MODEL`, `CLAUDE_DEFAULT_PROVIDER`), or per-node overrides.
+  Prompts support full `${{...}}` variable substitution; an optional
+  JSON-object response mode parses the output so downstream nodes can read
+  `${{nodeId.json.field}}`. Output also carries `model`, `provider`,
+  `stopReason`, and token `usage`. Runs through the engine's existing retry
+  and circuit-breaker machinery. New engine package `internal/llm`
+  (standard-library only, no vendor SDK).
+
 ## [0.1.3] — Patch release
 
 Two improvements that surfaced while validating v0.1.2 against a clean

CLAUDE.md

@@ -80,7 +80,7 @@ Real-time execution updates: browser connects directly to Go engine at `ws://hos
 ### Flow Editor
 
 - `src/components/flow/FlowCanvas.tsx` — React Flow surface (uses `reactflow` v11).
-- `src/components/flow/nodes/` — 23 node types built on shared `BaseNode` with `iconMap`/`colorMap`. Adding a node = component in `nodes/`, register in `nodes/index.ts`, properties editor in `properties/`, and (if the engine should execute it) a case in `internal/executor/flow_executor.go`.
+- `src/components/flow/nodes/` — 25 node types built on shared `BaseNode` with `iconMap`/`colorMap`. Adding a node = component in `nodes/`, register in `nodes/index.ts`, properties editor in `properties/`, and (if the engine should execute it) a case in `internal/executor/flow_executor.go`. The AI node (LLM completion) is dispatched to `internal/llm` from `flow_executor.go`.
 - `src/components/flow/properties/` — Per-node-type property editors.
 - `src/components/Combobox.tsx` — In-house searchable dropdown (no third-party lib). Used everywhere instead of native `<select>`.
 

README.md

@@ -4,7 +4,7 @@
 
 **Open-source workflow engine for self-hosters.**
 
-Drag-drop DAGs · cron schedules · 4 queue backends · code execution in 6 languages · pub/sub channels · AGPL.
+Drag-drop DAGs · **AI/LLM nodes** · cron schedules · 4 queue backends · code execution in 6 languages · pub/sub channels · AGPL.
 
 [![CI](https://github.com/EnterpriseX-Platform/RunLoop/actions/workflows/ci.yml/badge.svg)](https://github.com/EnterpriseX-Platform/RunLoop/actions/workflows/ci.yml)
 [![Release](https://img.shields.io/github/v/release/EnterpriseX-Platform/RunLoop?logo=github)](https://github.com/EnterpriseX-Platform/RunLoop/releases)
@@ -21,9 +21,10 @@ Drag-drop DAGs · cron schedules · 4 queue backends · code execution in 6 lang
   <img src="docs/screenshots/flow-editor.png" alt="RunLoop flow editor — drag-drop DAG with HTTP/DB/Shell/Python/Slack/Email node palette" width="900" />
 </p>
 
-Visual DAG editor. Multi-runtime code execution. Pluggable queue backends.
-Encrypted secret vault. WebSocket-streamed executions. Project-scoped
-multi-tenant. All in three containers, under 140 MB RAM idle, AGPL-3.0.
+Visual DAG editor. Multi-runtime code execution. In-flow AI/LLM nodes.
+Pluggable queue backends. Encrypted secret vault. WebSocket-streamed
+executions. Project-scoped multi-tenant. All in three containers, under
+140 MB RAM idle, AGPL-3.0.
 
 ```
 ┌──────────────┐   drag-drop    ┌──────────────┐   gocron + pg/redis/kafka  ┌─────────────┐
@@ -60,6 +61,7 @@ real-time over WebSocket.
 |---|---|---|---|---|---|
 | Drag-drop visual DAG editor | ✅ | ❌ form-based | ❌ | ❌ Python-defined | ✅ |
 | Multi-runtime code execution<br/>(Python · Node · Shell · Docker) | ✅ all 4 | ⚠️ via plugins | Shell only | Python-first | ✅ via plugins |
+| AI / LLM node in a step<br/>(Claude · OpenAI · Kimi) | ✅ built-in | ❌ | ❌ | ⚠️ provider pkg + code | ❌ |
 | Pluggable queue backends<br/>(Postgres · RabbitMQ · Kafka · Redis) | ✅ 4 backends | ❌ | ❌ | ⚠️ via Celery only | ❌ |
 | Built-in AES-256-GCM secret vault | ✅ | ❌ | ❌ | ⚠️ Connections (Fernet) | ❌ DB-stored |
 | WebSocket per-node execution stream | ✅ | ⚠️ live log tail | ❌ | ❌ poll | ❌ poll |
@@ -71,11 +73,26 @@ real-time over WebSocket.
 
 ## Features
 
-- **24 built-in node types** — Start/End, Condition, Switch, Loop (for-each / batch / parallel),
+- **25 built-in node types** — Start/End, Condition, Switch, Loop (for-each / batch / parallel),
   Transform, Merge, Delay, Set Variable, Sub-flow, Log, HTTP, Database (Postgres/MySQL),
   Shell, Python, Node.js, Docker, Slack, Email (SMTP), Webhook (signed outbound),
   Wait Webhook (inbound park), Enqueue (push to a queue), Notify (publish to a channel),
+  AI (LLM completion — Claude / OpenAI / Kimi, with JSON-mode output),
   Plugin (custom user-defined).
+- **🤖 AI node — drop an LLM into any flow, no glue code.** Summarize a
+  payload, classify an incoming webhook, extract fields, or draft a reply in a
+  single node. Pick Claude, OpenAI, or Kimi (keys live in the same encrypted
+  vault as the assistant — never in the flow), template the prompt with
+  upstream data (`${{input.body}}`), and flip on **JSON mode** to get
+  structured output the next node can branch on:
+
+  ```
+  Webhook ─▶ AI node ─▶ Switch ─┬─▶ Slack   (severity == "high")
+            classify to JSON    └─▶ Queue   (everything else)
+  ```
+
+  Token usage rides on every result, and the call retries through the same
+  circuit-breaker as every other node. No model SDK, no extra service.
 - **Variable substitution everywhere** — `${{nodeId.field}}`, `${{input.X}}`,
   `${{env.X}}`, `${{secrets.X}}`, plus `${{NOW}}` / `${{TODAY}}` and
   `${{loop.item}}` inside loop bodies.
@@ -221,6 +238,9 @@ For deeper dives see [`CLAUDE.md`](./CLAUDE.md) and [`docs/development/SETUP.md`
 ## Use cases we've seen
 
 - **Cron → external API → Slack alert** — three nodes, two minutes.
+- **AI triage** — Inbound webhook → **AI node** classifies the payload to JSON
+  (`{ "severity": "high", "team": "billing" }`) → Switch routes high-severity
+  to a Slack alert and the rest to a queue. Five nodes, no model SDK.
 - **Postgres ETL** — DB query → Transform → DB upsert, on a schedule, with
   a retry policy and a dead-letter queue.
 - **Webhook fan-out** — Inbound webhook (`WAIT_WEBHOOK`) parks until your

apps/runloop-engine/internal/executor/flow_executor.go

@@ -23,6 +23,7 @@ import (
 	"github.com/runloop/runloop-engine/internal/connector"
 	"github.com/runloop/runloop-engine/internal/db"
 	"github.com/runloop/runloop-engine/internal/idgen"
+	"github.com/runloop/runloop-engine/internal/llm"
 	"github.com/runloop/runloop-engine/internal/logging"
 	"github.com/runloop/runloop-engine/internal/models"
 	"github.com/runloop/runloop-engine/internal/worker"
@@ -659,6 +660,9 @@ func (fe *FlowExecutor) dispatchNode(
 
 	case "SLACK", "EMAIL", "WEBHOOK":
 		return fe.executeConnectorNode(ctx, string(node.Type), config)
+
+	case models.JobTypeAI:
+		return fe.executeAINode(ctx, task, config)
 	}
 
 	// Fall through to plugin registry. The node's type string becomes the
@@ -1438,6 +1442,186 @@ func (fe *FlowExecutor) executeConnectorNode(ctx context.Context, connType strin
 	}, nil
 }
 
+// aiModelSecret maps a provider to the secret name that holds its default
+// model override, mirroring the Next.js assistant proxy.
+var aiModelSecret = map[llm.Provider]string{
+	llm.ProviderClaude: "CLAUDE_DEFAULT_MODEL",
+	llm.ProviderOpenAI: "OPENAI_DEFAULT_MODEL",
+	llm.ProviderKimi:   "KIMI_DEFAULT_MODEL",
+}
+
+var aiKeySecret = map[llm.Provider]string{
+	llm.ProviderClaude: "CLAUDE_API_KEY",
+	llm.ProviderOpenAI: "OPENAI_API_KEY",
+	llm.ProviderKimi:   "KIMI_API_KEY",
+}
+
+// executeAINode runs an AI/LLM completion node. Config (post substitution):
+//
+//	prompt        string  required — the user message; ${{...}} already resolved
+//	systemPrompt  string  optional — system steer
+//	provider      string  optional — claude | openai | kimi (else auto-detect)
+//	model         string  optional — overrides the provider default
+//	apiKey        string  optional — explicit key (e.g. ${{secrets.X}}); else vault
+//	maxTokens     number  optional — capped at llm.MaxTokensCap
+//	temperature   number  optional
+//	responseFormat string optional — "json" asks for a single JSON object
+//	timeout       number  optional — seconds; default 60
+//
+// Output: response, model, provider, stopReason, usage{...}, and (json mode)
+// a parsed `json` object. Reference downstream as ${{nodeId.response}},
+// ${{nodeId.json.field}}, ${{nodeId.usage.totalTokens}}.
+func (fe *FlowExecutor) executeAINode(ctx context.Context, task *worker.Task, config models.JSONMap) (*models.JobResult, error) {
+	prompt, _ := config["prompt"].(string)
+	if strings.TrimSpace(prompt) == "" {
+		if m, ok := config["message"].(string); ok {
+			prompt = m
+		}
+	}
+	if strings.TrimSpace(prompt) == "" {
+		return &models.JobResult{
+			Success:      false,
+			ErrorMessage: strPtr("AI node: 'prompt' is required"),
+		}, nil
+	}
+
+	getSecret := func(name string) string {
+		if fe.secretStore == nil {
+			return ""
+		}
+		v, err := fe.secretStore.GetSecret(ctx, name, task.ProjectID)
+		if err != nil {
+			return ""
+		}
+		return v
+	}
+
+	// --- provider + key resolution (mirrors the assistant proxy) ---
+	keys := map[llm.Provider]string{
+		llm.ProviderClaude: getSecret(aiKeySecret[llm.ProviderClaude]),
+		llm.ProviderOpenAI: getSecret(aiKeySecret[llm.ProviderOpenAI]),
+		llm.ProviderKimi:   getSecret(aiKeySecret[llm.ProviderKimi]),
+	}
+
+	var provider llm.Provider
+	if p, _ := config["provider"].(string); strings.TrimSpace(p) != "" {
+		provider = llm.Provider(strings.ToLower(strings.TrimSpace(p)))
+	} else if pref := strings.ToLower(strings.TrimSpace(getSecret("CLAUDE_DEFAULT_PROVIDER"))); pref != "" {
+		provider = llm.Provider(pref)
+	}
+	switch provider {
+	case llm.ProviderClaude, llm.ProviderOpenAI, llm.ProviderKimi:
+		// explicit/preferred provider is valid
+	default:
+		// auto-detect: first provider with a configured key
+		switch {
+		case keys[llm.ProviderClaude] != "":
+			provider = llm.ProviderClaude
+		case keys[llm.ProviderOpenAI] != "":
+			provider = llm.ProviderOpenAI
+		case keys[llm.ProviderKimi] != "":
+			provider = llm.ProviderKimi
+		default:
+			return &models.JobResult{
+				Success:      false,
+				ErrorMessage: strPtr("AI node: no provider configured — set CLAUDE_API_KEY, OPENAI_API_KEY, or KIMI_API_KEY (or a node 'provider')"),
+			}, nil
+		}
+	}
+
+	apiKey, _ := config["apiKey"].(string)
+	if strings.TrimSpace(apiKey) == "" {
+		apiKey = keys[provider]
+	}
+	if strings.TrimSpace(apiKey) == "" {
+		return &models.JobResult{
+			Success:      false,
+			ErrorMessage: strPtr(fmt.Sprintf("AI node: provider %q selected but %s is not set in this project's secrets", provider, aiKeySecret[provider])),
+		}, nil
+	}
+
+	// --- model + tuning ---
+	model, _ := config["model"].(string)
+	if strings.TrimSpace(model) == "" {
+		model = getSecret(aiModelSecret[provider])
+	}
+
+	maxTokens := 1024
+	if v, ok := config["maxTokens"].(float64); ok && v > 0 {
+		maxTokens = int(v)
+	}
+
+	var temperature *float64
+	if v, ok := config["temperature"].(float64); ok {
+		t := v
+		temperature = &t
+	}
+
+	jsonMode := false
+	if rf, _ := config["responseFormat"].(string); strings.EqualFold(rf, "json") {
+		jsonMode = true
+	}
+
+	timeout := 60 * time.Second
+	if v, ok := config["timeout"].(float64); ok && v > 0 {
+		timeout = time.Duration(v) * time.Second
+	}
+
+	system, _ := config["systemPrompt"].(string)
+	if system == "" {
+		system, _ = config["system"].(string)
+	}
+
+	client := llm.New(&http.Client{Timeout: timeout})
+	resp, err := client.Complete(ctx, llm.Request{
+		Provider:    provider,
+		APIKey:      apiKey,
+		Model:       model,
+		System:      system,
+		Messages:    []llm.Message{{Role: "user", Content: prompt}},
+		MaxTokens:   maxTokens,
+		Temperature: temperature,
+		JSONMode:    jsonMode,
+	})
+	if err != nil {
+		return &models.JobResult{
+			Success:      false,
+			ErrorMessage: strPtr(err.Error()),
+		}, nil
+	}
+
+	output := models.JSONMap{
+		"response":   resp.Text,
+		"model":      resp.Model,
+		"provider":   string(resp.Provider),
+		"stopReason": resp.StopReason,
+		"usage": map[string]interface{}{
+			"promptTokens":     resp.Usage.PromptTokens,
+			"completionTokens": resp.Usage.CompletionTokens,
+			"totalTokens":      resp.Usage.TotalTokens,
+		},
+	}
+
+	// In JSON mode, surface the parsed object as `json` so downstream nodes can
+	// reference ${{nodeId.json.field}}. A parse failure isn't fatal — the raw
+	// text is still available on `response` — but we flag it in the logs.
+	parseNote := ""
+	if jsonMode {
+		var parsed interface{}
+		if err := json.Unmarshal([]byte(strings.TrimSpace(resp.Text)), &parsed); err == nil {
+			output["json"] = parsed
+		} else {
+			parseNote = " (responseFormat=json but output was not valid JSON)"
+		}
+	}
+
+	return &models.JobResult{
+		Success: true,
+		Output:  output,
+		Logs:    fmt.Sprintf("%s/%s · %d prompt + %d completion = %d tokens · stop=%s%s", resp.Provider, resp.Model, resp.Usage.PromptTokens, resp.Usage.CompletionTokens, resp.Usage.TotalTokens, resp.StopReason, parseNote),
+	}, nil
+}
+
 func (fe *FlowExecutor) resolveSecrets(ctx context.Context, config models.JSONMap, projectID string) (models.JSONMap, error) {
 	result := make(models.JSONMap)
 	for key, value := range config {