From ab06045cd224586e3477adf4e2501050c96a829a Mon Sep 17 00:00:00 2001
From: hacka0wi <124970567+hacka0wi@users.noreply.github.com>
Date: Tue, 9 Jun 2026 17:29:59 +0700
Subject: [PATCH 1/2] =?UTF-8?q?feat(flow):=20add=20AI=20node=20=E2=80=94?=
 =?UTF-8?q?=20in-flow=20LLM=20completion=20(Claude/OpenAI/Kimi)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds a 25th flow node that calls an LLM 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.

Engine:
- new internal/llm package — provider-agnostic client (Anthropic Messages
  API + OpenAI-compatible chat completions), standard-library only, no
  vendor SDK. Endpoints are vars so tests can target httptest.
- executeAINode + dispatch case models.JobTypeAI in flow_executor.go.
  Prompts are already ${{...}}-substituted by the time dispatch runs;
  optional JSON-object mode parses output to ${{nodeId.json.field}}.
  Output carries response/model/provider/stopReason/usage. Runs through
  the existing retry + circuit-breaker path.
- 5 new tests (provider request shaping, JSON mode, token cap, upstream
  error, validation).

UI:
- AINode + AINodeProperties, registered in nodes/properties indexes.
- new "AI" palette category (Sparkles icon, #8B5CF6), icon/color in BaseNode.
- node type 'ai' follows the existing nodeType.toUpperCase() === JobType
  convention, so it serializes to "AI" with no mapping changes.

Docs: README features (24→25 + AI bullet), CHANGELOG [Unreleased],
CLAUDE.md, in-app API docs node list.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  14 +
 CLAUDE.md                                     |   2 +-
 README.md                                     |   8 +-
 .../internal/executor/flow_executor.go        | 184 ++++++++++
 apps/runloop-engine/internal/llm/client.go    | 318 ++++++++++++++++++
 .../internal/llm/client_test.go               | 189 +++++++++++
 .../internal/models/scheduler.go              |   1 +
 .../(protected)/p/[projectId]/docs/page.tsx   |   2 +-
 .../src/components/flow/FlowCanvas.tsx        |   5 +-
 .../src/components/flow/nodes/AINode.tsx      |  26 ++
 .../src/components/flow/nodes/BaseNode.tsx    |   4 +-
 .../src/components/flow/nodes/index.ts        |   4 +-
 .../flow/properties/AINodeProperties.tsx      | 140 ++++++++
 .../src/components/flow/properties/index.ts   |   4 +-
 14 files changed, 894 insertions(+), 7 deletions(-)
 create mode 100644 apps/runloop-engine/internal/llm/client.go
 create mode 100644 apps/runloop-engine/internal/llm/client_test.go
 create mode 100644 apps/runloop/src/components/flow/nodes/AINode.tsx
 create mode 100644 apps/runloop/src/components/flow/properties/AINodeProperties.tsx

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 510d653..642fd45 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/CLAUDE.md b/CLAUDE.md
index 2868998..c165501 100644
--- a/CLAUDE.md
+++ b/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>`.
 
diff --git a/README.md b/README.md
index ef8dcf5..b1a5883 100644
--- a/README.md
+++ b/README.md
@@ -71,11 +71,17 @@ 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** — call Claude, OpenAI, or Kimi from inside a flow. Provider and
+  keys come from the same project vault as the in-app assistant; templated
+  prompts (`${{input.X}}`), optional JSON-object mode for structured output
+  (`${{nodeId.json.field}}`), and token usage on every result. Runs through
+  the engine's retry + circuit-breaker path like any other executor.
 - **Variable substitution everywhere** — `${{nodeId.field}}`, `${{input.X}}`,
   `${{env.X}}`, `${{secrets.X}}`, plus `${{NOW}}` / `${{TODAY}}` and
   `${{loop.item}}` inside loop bodies.
diff --git a/apps/runloop-engine/internal/executor/flow_executor.go b/apps/runloop-engine/internal/executor/flow_executor.go
index c579606..17e7670 100644
--- a/apps/runloop-engine/internal/executor/flow_executor.go
+++ b/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 {
diff --git a/apps/runloop-engine/internal/llm/client.go b/apps/runloop-engine/internal/llm/client.go
new file mode 100644
index 0000000..1b2cc23
--- /dev/null
+++ b/apps/runloop-engine/internal/llm/client.go
@@ -0,0 +1,318 @@
+// Package llm is a tiny, dependency-free client for the three chat providers
+// RunLoop supports: Claude (Anthropic Messages API), OpenAI, and Kimi
+// (Moonshot, OpenAI-compatible). It is the engine-side twin of the Next.js
+// AI assistant proxy (apps/runloop/src/app/api/ai/chat/route.ts) — same
+// provider set, same secret names, same default models — so a flow's AI node
+// behaves identically to the in-app assistant.
+//
+// The package deliberately depends only on the standard library: the AI node
+// is on the flow execution hot path and we don't want a vendor SDK pulling in
+// its own retry/transport behavior underneath the engine's own retry +
+// circuit-breaker machinery.
+package llm
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+// Provider identifies a chat backend. Values match the strings stored in the
+// CLAUDE_DEFAULT_PROVIDER secret and accepted by the AI node's `provider`
+// field.
+type Provider string
+
+const (
+	ProviderClaude Provider = "claude"
+	ProviderOpenAI Provider = "openai"
+	ProviderKimi   Provider = "kimi"
+)
+
+// DefaultModel is the per-provider fallback model, used when neither the node
+// config nor a <PROVIDER>_DEFAULT_MODEL secret specifies one. Kept in sync
+// with DEFAULT_MODEL in the Next.js route.
+var DefaultModel = map[Provider]string{
+	ProviderClaude: "claude-sonnet-4-7",
+	ProviderOpenAI: "gpt-4o-mini",
+	ProviderKimi:   "kimi-latest",
+}
+
+// MaxTokensCap is the hard ceiling on completion length, matching the
+// assistant proxy. Defends against a runaway flow burning a provider budget.
+const MaxTokensCap = 4096
+
+// Provider endpoints. Vars (not consts) so tests can point them at an
+// httptest server; never reassigned in production code.
+var (
+	anthropicEndpoint = "https://api.anthropic.com/v1/messages"
+	openaiEndpoint    = "https://api.openai.com/v1/chat/completions"
+	kimiEndpoint      = "https://api.moonshot.cn/v1/chat/completions"
+)
+
+// Message is one turn of the conversation. The AI node only ever sends a
+// single user message today, but the slice keeps the door open for an
+// agent/tool loop later.
+type Message struct {
+	Role    string `json:"role"` // "user" | "assistant"
+	Content string `json:"content"`
+}
+
+// Request is a provider-agnostic completion request.
+type Request struct {
+	Provider    Provider
+	APIKey      string
+	Model       string
+	System      string
+	Messages    []Message
+	MaxTokens   int
+	Temperature *float64 // pointer so "unset" is distinguishable from 0
+	JSONMode    bool     // ask the model to emit a single JSON object
+}
+
+// Usage reports token counts, normalized across providers.
+type Usage struct {
+	PromptTokens     int `json:"promptTokens"`
+	CompletionTokens int `json:"completionTokens"`
+	TotalTokens      int `json:"totalTokens"`
+}
+
+// Response is the normalized result.
+type Response struct {
+	Text       string   `json:"text"`
+	Model      string   `json:"model"`
+	Provider   Provider `json:"provider"`
+	StopReason string   `json:"stopReason"`
+	Usage      Usage    `json:"usage"`
+}
+
+// Client performs completions. The zero value is unusable; construct with New.
+type Client struct {
+	http *http.Client
+}
+
+// New builds a Client over the given http.Client. Pass an *http.Client with
+// the timeout you want the LLM call bounded by — the engine sizes this from
+// the node's timeout config.
+func New(httpClient *http.Client) *Client {
+	if httpClient == nil {
+		httpClient = http.DefaultClient
+	}
+	return &Client{http: httpClient}
+}
+
+// Complete dispatches to the right provider and returns a normalized response.
+func (c *Client) Complete(ctx context.Context, req Request) (*Response, error) {
+	if req.APIKey == "" {
+		return nil, fmt.Errorf("llm: API key is empty for provider %q", req.Provider)
+	}
+	if len(req.Messages) == 0 {
+		return nil, fmt.Errorf("llm: at least one message is required")
+	}
+	if req.MaxTokens <= 0 {
+		req.MaxTokens = 1024
+	}
+	if req.MaxTokens > MaxTokensCap {
+		req.MaxTokens = MaxTokensCap
+	}
+	if req.Model == "" {
+		req.Model = DefaultModel[req.Provider]
+	}
+
+	switch req.Provider {
+	case ProviderClaude:
+		return c.completeClaude(ctx, req)
+	case ProviderOpenAI:
+		return c.completeOpenAICompatible(ctx, req, openaiEndpoint)
+	case ProviderKimi:
+		return c.completeOpenAICompatible(ctx, req, kimiEndpoint)
+	default:
+		return nil, fmt.Errorf("llm: unknown provider %q", req.Provider)
+	}
+}
+
+// ---- Anthropic (Claude) ----
+
+func (c *Client) completeClaude(ctx context.Context, req Request) (*Response, error) {
+	system := req.System
+	if req.JSONMode {
+		// Claude has no native JSON-mode param, so steer via the system text.
+		hint := "Respond with a single valid JSON object and nothing else — no prose, no markdown fences."
+		if system == "" {
+			system = hint
+		} else {
+			system = system + "\n\n" + hint
+		}
+	}
+
+	payload := map[string]interface{}{
+		"model":      req.Model,
+		"max_tokens": req.MaxTokens,
+		"messages":   req.Messages,
+	}
+	if system != "" {
+		payload["system"] = system
+	}
+	if req.Temperature != nil {
+		payload["temperature"] = *req.Temperature
+	}
+
+	httpReq, err := c.newJSONRequest(ctx, anthropicEndpoint, payload)
+	if err != nil {
+		return nil, err
+	}
+	httpReq.Header.Set("x-api-key", req.APIKey)
+	httpReq.Header.Set("anthropic-version", "2023-06-01")
+
+	body, err := c.do(httpReq, ProviderClaude)
+	if err != nil {
+		return nil, err
+	}
+
+	var parsed struct {
+		Model      string `json:"model"`
+		StopReason string `json:"stop_reason"`
+		Content    []struct {
+			Type string `json:"type"`
+			Text string `json:"text"`
+		} `json:"content"`
+		Usage struct {
+			InputTokens  int `json:"input_tokens"`
+			OutputTokens int `json:"output_tokens"`
+		} `json:"usage"`
+	}
+	if err := json.Unmarshal(body, &parsed); err != nil {
+		return nil, fmt.Errorf("llm: failed to parse Claude response: %w", err)
+	}
+
+	var text strings.Builder
+	for _, block := range parsed.Content {
+		if block.Type == "text" {
+			text.WriteString(block.Text)
+		}
+	}
+
+	return &Response{
+		Text:       text.String(),
+		Model:      orDefault(parsed.Model, req.Model),
+		Provider:   ProviderClaude,
+		StopReason: parsed.StopReason,
+		Usage: Usage{
+			PromptTokens:     parsed.Usage.InputTokens,
+			CompletionTokens: parsed.Usage.OutputTokens,
+			TotalTokens:      parsed.Usage.InputTokens + parsed.Usage.OutputTokens,
+		},
+	}, nil
+}
+
+// ---- OpenAI / Kimi (chat completions) ----
+
+func (c *Client) completeOpenAICompatible(ctx context.Context, req Request, endpoint string) (*Response, error) {
+	messages := make([]map[string]string, 0, len(req.Messages)+1)
+	if req.System != "" {
+		messages = append(messages, map[string]string{"role": "system", "content": req.System})
+	}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]string{"role": m.Role, "content": m.Content})
+	}
+
+	payload := map[string]interface{}{
+		"model":      req.Model,
+		"messages":   messages,
+		"max_tokens": req.MaxTokens,
+	}
+	if req.Temperature != nil {
+		payload["temperature"] = *req.Temperature
+	}
+	if req.JSONMode {
+		payload["response_format"] = map[string]string{"type": "json_object"}
+	}
+
+	httpReq, err := c.newJSONRequest(ctx, endpoint, payload)
+	if err != nil {
+		return nil, err
+	}
+	httpReq.Header.Set("Authorization", "Bearer "+req.APIKey)
+
+	body, err := c.do(httpReq, req.Provider)
+	if err != nil {
+		return nil, err
+	}
+
+	var parsed struct {
+		Model   string `json:"model"`
+		Choices []struct {
+			Message      struct{ Content string `json:"content"` } `json:"message"`
+			FinishReason string                                     `json:"finish_reason"`
+		} `json:"choices"`
+		Usage struct {
+			PromptTokens     int `json:"prompt_tokens"`
+			CompletionTokens int `json:"completion_tokens"`
+			TotalTokens      int `json:"total_tokens"`
+		} `json:"usage"`
+	}
+	if err := json.Unmarshal(body, &parsed); err != nil {
+		return nil, fmt.Errorf("llm: failed to parse %s response: %w", req.Provider, err)
+	}
+	if len(parsed.Choices) == 0 {
+		return nil, fmt.Errorf("llm: %s returned no choices", req.Provider)
+	}
+
+	return &Response{
+		Text:       parsed.Choices[0].Message.Content,
+		Model:      orDefault(parsed.Model, req.Model),
+		Provider:   req.Provider,
+		StopReason: parsed.Choices[0].FinishReason,
+		Usage: Usage{
+			PromptTokens:     parsed.Usage.PromptTokens,
+			CompletionTokens: parsed.Usage.CompletionTokens,
+			TotalTokens:      parsed.Usage.TotalTokens,
+		},
+	}, nil
+}
+
+// ---- shared transport ----
+
+func (c *Client) newJSONRequest(ctx context.Context, endpoint string, payload interface{}) (*http.Request, error) {
+	buf, err := json.Marshal(payload)
+	if err != nil {
+		return nil, fmt.Errorf("llm: failed to encode request: %w", err)
+	}
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(buf))
+	if err != nil {
+		return nil, fmt.Errorf("llm: failed to build request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	return httpReq, nil
+}
+
+func (c *Client) do(req *http.Request, provider Provider) ([]byte, error) {
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("llm: %s request failed: %w", provider, err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("llm: failed to read %s response: %w", provider, err)
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		snippet := string(body)
+		if len(snippet) > 500 {
+			snippet = snippet[:500]
+		}
+		return nil, fmt.Errorf("llm: %s API %d: %s", provider, resp.StatusCode, snippet)
+	}
+	return body, nil
+}
+
+func orDefault(v, fallback string) string {
+	if v == "" {
+		return fallback
+	}
+	return v
+}
diff --git a/apps/runloop-engine/internal/llm/client_test.go b/apps/runloop-engine/internal/llm/client_test.go
new file mode 100644
index 0000000..ffbbfce
--- /dev/null
+++ b/apps/runloop-engine/internal/llm/client_test.go
@@ -0,0 +1,189 @@
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestCompleteClaude(t *testing.T) {
+	var gotPath, gotKey, gotVersion string
+	var gotBody map[string]interface{}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		gotKey = r.Header.Get("x-api-key")
+		gotVersion = r.Header.Get("anthropic-version")
+		raw, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(raw, &gotBody)
+
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{
+			"model": "claude-sonnet-4-7",
+			"stop_reason": "end_turn",
+			"content": [{"type":"text","text":"hello world"}],
+			"usage": {"input_tokens": 12, "output_tokens": 3}
+		}`))
+	}))
+	defer srv.Close()
+
+	orig := anthropicEndpoint
+	anthropicEndpoint = srv.URL
+	defer func() { anthropicEndpoint = orig }()
+
+	temp := 0.5
+	resp, err := New(srv.Client()).Complete(context.Background(), Request{
+		Provider:    ProviderClaude,
+		APIKey:      "sk-test",
+		System:      "be terse",
+		Messages:    []Message{{Role: "user", Content: "hi"}},
+		MaxTokens:   100,
+		Temperature: &temp,
+	})
+	if err != nil {
+		t.Fatalf("Complete returned error: %v", err)
+	}
+
+	if gotPath == "" || gotKey != "sk-test" || gotVersion != "2023-06-01" {
+		t.Errorf("request headers wrong: key=%q version=%q", gotKey, gotVersion)
+	}
+	if gotBody["system"] != "be terse" {
+		t.Errorf("system not forwarded: %v", gotBody["system"])
+	}
+	if gotBody["max_tokens"].(float64) != 100 {
+		t.Errorf("max_tokens wrong: %v", gotBody["max_tokens"])
+	}
+	if gotBody["temperature"].(float64) != 0.5 {
+		t.Errorf("temperature wrong: %v", gotBody["temperature"])
+	}
+	if resp.Text != "hello world" {
+		t.Errorf("text = %q, want %q", resp.Text, "hello world")
+	}
+	if resp.StopReason != "end_turn" {
+		t.Errorf("stopReason = %q", resp.StopReason)
+	}
+	if resp.Usage.TotalTokens != 15 {
+		t.Errorf("totalTokens = %d, want 15", resp.Usage.TotalTokens)
+	}
+}
+
+func TestCompleteOpenAIJSONMode(t *testing.T) {
+	var gotBody map[string]interface{}
+	var gotAuth string
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotAuth = r.Header.Get("Authorization")
+		raw, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(raw, &gotBody)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{
+			"model": "gpt-4o-mini",
+			"choices": [{"message": {"content": "{\"ok\":true}"}, "finish_reason": "stop"}],
+			"usage": {"prompt_tokens": 5, "completion_tokens": 4, "total_tokens": 9}
+		}`))
+	}))
+	defer srv.Close()
+
+	orig := openaiEndpoint
+	openaiEndpoint = srv.URL
+	defer func() { openaiEndpoint = orig }()
+
+	resp, err := New(srv.Client()).Complete(context.Background(), Request{
+		Provider:  ProviderOpenAI,
+		APIKey:    "sk-oai",
+		System:    "sys",
+		Messages:  []Message{{Role: "user", Content: "give me json"}},
+		MaxTokens: 50,
+		JSONMode:  true,
+	})
+	if err != nil {
+		t.Fatalf("Complete returned error: %v", err)
+	}
+
+	if gotAuth != "Bearer sk-oai" {
+		t.Errorf("auth header = %q", gotAuth)
+	}
+	rf, ok := gotBody["response_format"].(map[string]interface{})
+	if !ok || rf["type"] != "json_object" {
+		t.Errorf("response_format not set for JSON mode: %v", gotBody["response_format"])
+	}
+	// system + user message should both be present, system first
+	msgs, _ := gotBody["messages"].([]interface{})
+	if len(msgs) != 2 {
+		t.Fatalf("expected 2 messages, got %d", len(msgs))
+	}
+	if first := msgs[0].(map[string]interface{}); first["role"] != "system" {
+		t.Errorf("first message role = %v, want system", first["role"])
+	}
+	if resp.Text != `{"ok":true}` {
+		t.Errorf("text = %q", resp.Text)
+	}
+	if resp.Usage.TotalTokens != 9 {
+		t.Errorf("totalTokens = %d, want 9", resp.Usage.TotalTokens)
+	}
+}
+
+func TestCompleteMaxTokensCapped(t *testing.T) {
+	var gotBody map[string]interface{}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		raw, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(raw, &gotBody)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"content":[{"type":"text","text":"x"}],"usage":{}}`))
+	}))
+	defer srv.Close()
+
+	orig := anthropicEndpoint
+	anthropicEndpoint = srv.URL
+	defer func() { anthropicEndpoint = orig }()
+
+	_, err := New(srv.Client()).Complete(context.Background(), Request{
+		Provider:  ProviderClaude,
+		APIKey:    "k",
+		Messages:  []Message{{Role: "user", Content: "hi"}},
+		MaxTokens: 999999,
+	})
+	if err != nil {
+		t.Fatalf("Complete returned error: %v", err)
+	}
+	if gotBody["max_tokens"].(float64) != float64(MaxTokensCap) {
+		t.Errorf("max_tokens = %v, want capped at %d", gotBody["max_tokens"], MaxTokensCap)
+	}
+}
+
+func TestCompleteUpstreamError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+		_, _ = w.Write([]byte(`{"error":"invalid key"}`))
+	}))
+	defer srv.Close()
+
+	orig := openaiEndpoint
+	openaiEndpoint = srv.URL
+	defer func() { openaiEndpoint = orig }()
+
+	_, err := New(srv.Client()).Complete(context.Background(), Request{
+		Provider: ProviderOpenAI,
+		APIKey:   "bad",
+		Messages: []Message{{Role: "user", Content: "hi"}},
+	})
+	if err == nil {
+		t.Fatal("expected error on 401, got nil")
+	}
+}
+
+func TestCompleteValidation(t *testing.T) {
+	c := New(nil)
+	if _, err := c.Complete(context.Background(), Request{Provider: ProviderClaude, Messages: []Message{{Role: "user", Content: "x"}}}); err == nil {
+		t.Error("expected error for empty API key")
+	}
+	if _, err := c.Complete(context.Background(), Request{Provider: ProviderClaude, APIKey: "k"}); err == nil {
+		t.Error("expected error for no messages")
+	}
+	if _, err := c.Complete(context.Background(), Request{Provider: "bogus", APIKey: "k", Messages: []Message{{Role: "user", Content: "x"}}}); err == nil {
+		t.Error("expected error for unknown provider")
+	}
+}
diff --git a/apps/runloop-engine/internal/models/scheduler.go b/apps/runloop-engine/internal/models/scheduler.go
index 512da33..ba36e38 100644
--- a/apps/runloop-engine/internal/models/scheduler.go
+++ b/apps/runloop-engine/internal/models/scheduler.go
@@ -51,6 +51,7 @@ const (
 	JobTypeDocker    JobType = "DOCKER"
 	JobTypeSlack     JobType = "SLACK"
 	JobTypeEmail     JobType = "EMAIL"
+	JobTypeAI        JobType = "AI"
 	// Flow-shape nodes: handled directly by FlowExecutor, not by per-type executors.
 	JobTypeStart     JobType = "START"
 	JobTypeEnd       JobType = "END"
diff --git a/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx b/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
index 3f6b013..b18958e 100644
--- a/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
+++ b/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
@@ -60,7 +60,7 @@ const RAW_ENDPOINTS: EndpointDef[] = [
   { group: 'Flows', method: 'POST', path: '/api/flows',
     summary: 'Create a flow. flowConfig has nodes[] + edges[].',
     body: '{\n  "projectId": "demo-project",\n  "name": "send-welcome",\n  "type": "DAG",\n  "flowConfig": { "edges": [...], "nodes": [...] }\n}',
-    notes: 'Node types: START · END · CONDITION · DELAY · LOOP · TRANSFORM · MERGE · SWITCH · LOG · SET_VARIABLE · SUBFLOW · WEBHOOK_OUT · WAIT_WEBHOOK · ENQUEUE · NOTIFY · HTTP · DATABASE · SHELL · PYTHON · NODEJS · DOCKER · SLACK · EMAIL.' },
+    notes: 'Node types: START · END · CONDITION · DELAY · LOOP · TRANSFORM · MERGE · SWITCH · LOG · SET_VARIABLE · SUBFLOW · WEBHOOK_OUT · WAIT_WEBHOOK · ENQUEUE · NOTIFY · HTTP · DATABASE · SHELL · PYTHON · NODEJS · DOCKER · SLACK · EMAIL · AI.' },
 
   // Channels
   { group: 'Channels', method: 'GET',  path: '/api/channels',                  summary: 'Active pub/sub channels with live subscriber counts.' },
diff --git a/apps/runloop/src/components/flow/FlowCanvas.tsx b/apps/runloop/src/components/flow/FlowCanvas.tsx
index 2d8e339..b2a6a4f 100644
--- a/apps/runloop/src/components/flow/FlowCanvas.tsx
+++ b/apps/runloop/src/components/flow/FlowCanvas.tsx
@@ -24,7 +24,7 @@ import 'reactflow/dist/style.css';
 import {
   Play, Square, Globe, Database, Terminal, Code2, Hash,
   Container as Docker, Slack, Mail, GitBranch, Clock, RotateCcw, Variable,
-  GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing,
+  GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing, Sparkles,
   Save, TestTube2, Wand2, Plus, Settings, X, AlertCircle, Search, ChevronDown,
 } from 'lucide-react';
 import { nodeTypes as importedNodeTypes } from './nodes';
@@ -51,6 +51,9 @@ const PALETTE = [
     { type: 'nodejsNode', label: 'Node.js', nodeType: 'nodejs', icon: Hash, color: '#22C55E' },
     { type: 'dockerNode', label: 'Docker', nodeType: 'docker', icon: Docker, color: '#0EA5E9' },
   ]},
+  { category: 'AI', items: [
+    { type: 'aiNode', label: 'AI', nodeType: 'ai', icon: Sparkles, color: '#8B5CF6' },
+  ]},
   { category: 'Notifications', items: [
     { type: 'slackNode', label: 'Slack', nodeType: 'slack', icon: Slack, color: '#A855F7' },
     { type: 'emailNode', label: 'Email', nodeType: 'email', icon: Mail, color: '#F97316' },
diff --git a/apps/runloop/src/components/flow/nodes/AINode.tsx b/apps/runloop/src/components/flow/nodes/AINode.tsx
new file mode 100644
index 0000000..24708e5
--- /dev/null
+++ b/apps/runloop/src/components/flow/nodes/AINode.tsx
@@ -0,0 +1,26 @@
+'use client';
+
+import { NodeProps } from 'reactflow';
+import { BaseNode, BaseNodeData } from './BaseNode';
+
+const PROVIDER_LABEL: Record<string, string> = {
+  claude: 'Claude',
+  openai: 'OpenAI',
+  kimi: 'Kimi',
+};
+
+export function AINode(props: NodeProps<BaseNodeData>) {
+  const { provider, model } = props.data.config || {};
+  const label = provider ? PROVIDER_LABEL[provider] || provider : 'auto';
+
+  return (
+    <BaseNode {...props} inputs={1} outputs={1}>
+      <div className="flex items-center gap-2">
+        <span className="text-xs font-medium text-[#8B5CF6]">{label}</span>
+        {model && (
+          <span className="text-xs text-[#a1a1aa] truncate max-w-[160px]">{model}</span>
+        )}
+      </div>
+    </BaseNode>
+  );
+}
diff --git a/apps/runloop/src/components/flow/nodes/BaseNode.tsx b/apps/runloop/src/components/flow/nodes/BaseNode.tsx
index d924365..f06affb 100644
--- a/apps/runloop/src/components/flow/nodes/BaseNode.tsx
+++ b/apps/runloop/src/components/flow/nodes/BaseNode.tsx
@@ -5,7 +5,7 @@ import { Handle, Position, NodeProps } from 'reactflow';
 import {
   Play, Square, Globe, Database, Terminal, Code2, Hash,
   Slack, Mail, GitBranch, Clock, RotateCcw, Variable, Container,
-  GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing,
+  GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing, Sparkles,
 } from 'lucide-react';
 
 export interface BaseNodeData {
@@ -40,6 +40,7 @@ const iconMap: Record<string, React.ElementType> = {
   wait_webhook: Hourglass,
   enqueue: Inbox,
   notify: BellRing,
+  ai: Sparkles,
 };
 
 const colorMap: Record<string, { hex: string; label: string }> = {
@@ -66,6 +67,7 @@ const colorMap: Record<string, { hex: string; label: string }> = {
   wait_webhook: { hex: '#FB923C', label: 'Wait Webhook' },
   enqueue:      { hex: '#0EA5E9', label: 'Enqueue' },
   notify:       { hex: '#F59E0B', label: 'Notify' },
+  ai:           { hex: '#8B5CF6', label: 'AI' },
 };
 
 const MONO = "'IBM Plex Mono', ui-monospace, SFMono-Regular, Menlo, monospace";
diff --git a/apps/runloop/src/components/flow/nodes/index.ts b/apps/runloop/src/components/flow/nodes/index.ts
index 070b8be..cd3d2e9 100644
--- a/apps/runloop/src/components/flow/nodes/index.ts
+++ b/apps/runloop/src/components/flow/nodes/index.ts
@@ -21,13 +21,14 @@ import { WebhookNode } from './WebhookNode';
 import { WaitWebhookNode } from './WaitWebhookNode';
 import { EnqueueNode } from './EnqueueNode';
 import { NotifyNode } from './NotifyNode';
+import { AINode } from './AINode';
 import { PluginNode } from './PluginNode';
 
 export {
   StartNode, EndNode, HttpNode, DatabaseNode, ShellNode, PythonNode,
   NodejsNode, DockerNode, SlackNode, EmailNode, ConditionNode, DelayNode,
   LoopNode, TransformNode, MergeNode, SwitchNode,
-  LogNode, SetVarNode, SubFlowNode, WebhookNode, WaitWebhookNode, EnqueueNode, NotifyNode, PluginNode,
+  LogNode, SetVarNode, SubFlowNode, WebhookNode, WaitWebhookNode, EnqueueNode, NotifyNode, AINode, PluginNode,
 };
 
 export const nodeTypes = {
@@ -54,5 +55,6 @@ export const nodeTypes = {
   waitWebhookNode: WaitWebhookNode,
   enqueueNode: EnqueueNode,
   notifyNode: NotifyNode,
+  aiNode: AINode,
   pluginNode: PluginNode,
 };
diff --git a/apps/runloop/src/components/flow/properties/AINodeProperties.tsx b/apps/runloop/src/components/flow/properties/AINodeProperties.tsx
new file mode 100644
index 0000000..6b04b26
--- /dev/null
+++ b/apps/runloop/src/components/flow/properties/AINodeProperties.tsx
@@ -0,0 +1,140 @@
+'use client';
+
+import {
+  BaseProperties,
+  Section,
+  TextField,
+  TextArea,
+  SelectField,
+  NumberField,
+  SecretField,
+  VariableHint,
+} from './BaseProperties';
+import { BasePropertiesProps } from './BaseProperties';
+
+const PROVIDERS = [
+  { value: '', label: 'Auto (project default)' },
+  { value: 'claude', label: 'Claude (Anthropic)' },
+  { value: 'openai', label: 'OpenAI (ChatGPT)' },
+  { value: 'kimi', label: 'Kimi (Moonshot)' },
+];
+
+const RESPONSE_FORMATS = [
+  { value: 'text', label: 'Text' },
+  { value: 'json', label: 'JSON object' },
+];
+
+const MODEL_PLACEHOLDER: Record<string, string> = {
+  claude: 'claude-sonnet-4-7',
+  openai: 'gpt-4o-mini',
+  kimi: 'kimi-latest',
+  '': 'provider default',
+};
+
+export function AINodeProperties({ config, onChange }: BasePropertiesProps) {
+  const provider = config.provider || '';
+  const jsonMode = (config.responseFormat || 'text') === 'json';
+
+  return (
+    <BaseProperties config={config} onChange={onChange}>
+      <Section title="Model">
+        <div className="space-y-4">
+          <SelectField
+            label="Provider"
+            value={provider}
+            onChange={(v) => onChange({ ...config, provider: v })}
+            options={PROVIDERS}
+          />
+          <p className="text-xs -mt-2" style={{ color: 'var(--t-text-muted)' }}>
+            Keys come from the secret vault (<code style={{ color: 'var(--t-accent)' }}>CLAUDE_API_KEY</code>,{' '}
+            <code style={{ color: 'var(--t-accent)' }}>OPENAI_API_KEY</code>,{' '}
+            <code style={{ color: 'var(--t-accent)' }}>KIMI_API_KEY</code>). Auto picks your project default.
+          </p>
+          <TextField
+            label="Model (optional)"
+            value={config.model || ''}
+            onChange={(v) => onChange({ ...config, model: v })}
+            placeholder={MODEL_PLACEHOLDER[provider] || 'provider default'}
+          />
+        </div>
+      </Section>
+
+      <Section title="Prompt">
+        <div className="space-y-4">
+          <TextArea
+            label="System Prompt (optional)"
+            value={config.systemPrompt || ''}
+            onChange={(v) => onChange({ ...config, systemPrompt: v })}
+            placeholder="You are a concise assistant that summarizes incident reports."
+            rows={3}
+          />
+          <TextArea
+            label="Prompt"
+            value={config.prompt || ''}
+            onChange={(v) => onChange({ ...config, prompt: v })}
+            placeholder={'Summarize this payload in one sentence:\n${{input.body}}'}
+            rows={6}
+          />
+          <VariableHint />
+        </div>
+      </Section>
+
+      <Section title="Output & Tuning">
+        <div className="space-y-4">
+          <SelectField
+            label="Response Format"
+            value={config.responseFormat || 'text'}
+            onChange={(v) => onChange({ ...config, responseFormat: v })}
+            options={RESPONSE_FORMATS}
+          />
+          {jsonMode && (
+            <p className="text-xs -mt-2" style={{ color: 'var(--t-text-muted)' }}>
+              The model is asked to return one JSON object. Reference fields downstream as{' '}
+              <code style={{ color: 'var(--t-accent)' }}>{'${{nodeId.json.field}}'}</code>.
+            </p>
+          )}
+          <div className="grid grid-cols-2 gap-4">
+            <NumberField
+              label="Max Tokens"
+              value={config.maxTokens ?? 1024}
+              onChange={(v) => onChange({ ...config, maxTokens: v })}
+              min={1}
+              max={4096}
+              step={64}
+            />
+            <NumberField
+              label="Temperature"
+              value={config.temperature ?? 0.7}
+              onChange={(v) => onChange({ ...config, temperature: v })}
+              min={0}
+              max={2}
+              step={0.1}
+            />
+          </div>
+          <NumberField
+            label="Timeout (seconds)"
+            value={config.timeout ?? 60}
+            onChange={(v) => onChange({ ...config, timeout: v })}
+            min={1}
+            max={600}
+          />
+        </div>
+      </Section>
+
+      <Section title="Advanced">
+        <div className="space-y-4">
+          <SecretField
+            label="API Key Override (optional)"
+            value={config.apiKey || ''}
+            onChange={(v) => onChange({ ...config, apiKey: v })}
+            placeholder="{{secrets.CLAUDE_API_KEY}}"
+            suggestions={['CLAUDE_API_KEY', 'OPENAI_API_KEY', 'KIMI_API_KEY']}
+          />
+          <p className="text-xs -mt-2" style={{ color: 'var(--t-text-muted)' }}>
+            Leave empty to use the provider key from the vault automatically.
+          </p>
+        </div>
+      </Section>
+    </BaseProperties>
+  );
+}
diff --git a/apps/runloop/src/components/flow/properties/index.ts b/apps/runloop/src/components/flow/properties/index.ts
index dce0bdc..fca92ef 100644
--- a/apps/runloop/src/components/flow/properties/index.ts
+++ b/apps/runloop/src/components/flow/properties/index.ts
@@ -21,6 +21,7 @@ import { WebhookNodeProperties } from './WebhookNodeProperties';
 import { WaitWebhookNodeProperties } from './WaitWebhookNodeProperties';
 import { EnqueueNodeProperties } from './EnqueueNodeProperties';
 import { NotifyNodeProperties } from './NotifyNodeProperties';
+import { AINodeProperties } from './AINodeProperties';
 
 export {
   HttpNodeProperties, DatabaseNodeProperties, SlackNodeProperties, EmailNodeProperties,
@@ -28,7 +29,7 @@ export {
   PythonNodeProperties, NodejsNodeProperties, DockerNodeProperties, S3NodeProperties, RedisNodeProperties,
   MergeNodeProperties, SwitchNodeProperties,
   LogNodeProperties, SetVarNodeProperties, SubFlowNodeProperties,
-  WebhookNodeProperties, WaitWebhookNodeProperties, EnqueueNodeProperties, NotifyNodeProperties,
+  WebhookNodeProperties, WaitWebhookNodeProperties, EnqueueNodeProperties, NotifyNodeProperties, AINodeProperties,
 };
 
 export const propertiesComponents: Record<string, React.ComponentType<any>> = {
@@ -54,4 +55,5 @@ export const propertiesComponents: Record<string, React.ComponentType<any>> = {
   wait_webhook: WaitWebhookNodeProperties,
   enqueue: EnqueueNodeProperties,
   notify: NotifyNodeProperties,
+  ai: AINodeProperties,
 };

From 919b5508d37375ddab86ca27549689e5cf31ffea Mon Sep 17 00:00:00 2001
From: hacka0wi <124970567+hacka0wi@users.noreply.github.com>
Date: Tue, 9 Jun 2026 17:43:58 +0700
Subject: [PATCH 2/2] docs(readme): make the AI node a headline feature, not a
 footnote
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Reposition the AI node for adoption: lead the hero tagline + paragraph with
AI/LLM nodes, add an "AI / LLM node in a step" row to the comparison table
(built-in vs none of Cronicle/Dkron/DolphinScheduler; Airflow only via a
provider pkg + code), rewrite the feature bullet as benefit-driven with a
concrete webhook→classify→route example, and add an "AI triage" use case.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md | 32 +++++++++++++++++++++++---------
 1 file changed, 23 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index b1a5883..b0aa00c 100644
--- a/README.md
+++ b/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 |
@@ -77,11 +79,20 @@ real-time over WebSocket.
   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** — call Claude, OpenAI, or Kimi from inside a flow. Provider and
-  keys come from the same project vault as the in-app assistant; templated
-  prompts (`${{input.X}}`), optional JSON-object mode for structured output
-  (`${{nodeId.json.field}}`), and token usage on every result. Runs through
-  the engine's retry + circuit-breaker path like any other executor.
+- **🤖 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.
@@ -227,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