uopi/internal/controllogic/engine.go

package controllogic

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/uopi/uopi/internal/audit"
	"github.com/uopi/uopi/internal/broker"
	"github.com/uopi/uopi/internal/confmgr"
)

// errUnknownSource is returned by config-apply writes targeting a data source
// that isn't registered with the broker.
var errUnknownSource = errors.New("unknown data source")

// formatAny renders a config value for the audit log.
func formatAny(v any) string {
	switch x := v.(type) {
	case float64:
		return strconv.FormatFloat(x, 'g', -1, 64)
	case string:
		return x
	default:
		return fmt.Sprintf("%v", v)
	}
}

// Guards against runaway flows (cycles / pathological loops).
const (
	maxSteps = 100000
	maxLoop  = 100000
)

// Engine runs all enabled control-logic graphs continuously under a root
// context. Reload tears down the current generation (subscriptions, timers,
// in-flight flows) and rebuilds from the store's enabled graphs.
type Engine struct {
	broker *broker.Broker
	store  *Store
	cfg    *confmgr.Store
	audit  audit.Recorder
	log    *slog.Logger
	root   context.Context

	mu     sync.Mutex
	cancel context.CancelFunc // cancels the current generation
	wg     *sync.WaitGroup    // tracks the current generation's goroutines

	// notifier delivers action.dialog requests to connected clients. Stored in
	// an atomic so flow goroutines can read it without taking e.mu (which Reload
	// holds while it waits for those same goroutines to drain).
	notifier atomic.Value // notifierBox

	// debugObs receives per-node execution events for the live debug view, and
	// debugWatch is the set of graph ids with at least one live subscriber. Both
	// are read lock-free by flow goroutines (same trick as notifier); emitDebug
	// short-circuits cheaply when the firing graph has no watcher.
	debugObs   atomic.Value // debugObsBox
	debugWatch atomic.Value // map[string]bool (immutable snapshot)

	// fireChs maps a debug route id (live graph id or simulate sandbox id) to its
	// compiled graph's manual-fire channel, so the debug UI can force a trigger to
	// run. Entries are added/removed as generations (and simulate sandboxes) come
	// and go; FireTrigger does a non-blocking send so a torn-down graph drops.
	fireMu  sync.Mutex
	fireChs map[string]chan string

	// Shared live signal cache for the current generation (key "ds\0name").
	liveMu sync.RWMutex
	live   map[string]float64
}

// notifierBox wraps a Notifier so atomic.Value always sees one concrete type.
type notifierBox struct{ n Notifier }

// dialogSeq generates unique action.dialog ids across all graphs.
var dialogSeq uint64

// SetNotifier installs the sink for action.dialog requests. Safe to call once
// at startup before or after Reload; it is read lock-free by running flows.
func (e *Engine) SetNotifier(n Notifier) {
	e.notifier.Store(notifierBox{n: n})
}

// NewEngine creates an engine bound to root. Call Reload to start it. rec records
// the writes performed by flows; pass audit.Nop() to disable auditing.
func NewEngine(root context.Context, brk *broker.Broker, store *Store, cfg *confmgr.Store, rec audit.Recorder, log *slog.Logger) *Engine {
	if rec == nil {
		rec = audit.Nop()
	}
	return &Engine{
		broker:  brk,
		store:   store,
		cfg:     cfg,
		audit:   rec,
		log:     log,
		root:    root,
		live:    map[string]float64{},
		fireChs: map[string]chan string{},
	}
}

// ── reference / value helpers ──────────────────────────────────────────────────

func refKey(ds, name string) string { return ds + "\x00" + name }

// parseRef splits a "ds:name" target on the FIRST ':' (EPICS PV names contain
// ':'). A bare name (no ':') is a graph-local variable in data source "local".
func parseRef(target string) (ds, name string, ok bool) {
	t := strings.TrimSpace(target)
	if t == "" {
		return "", "", false
	}
	i := strings.IndexByte(t, ':')
	if i < 0 {
		return "local", t, true
	}
	return t[:i], t[i+1:], true
}

func toNum(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case float32:
		return float64(x)
	case int64:
		return float64(x)
	case int:
		return float64(x)
	case bool:
		if x {
			return 1
		}
		return 0
	case string:
		f, err := strconv.ParseFloat(strings.TrimSpace(x), 64)
		if err != nil {
			return math.NaN()
		}
		return f
	default:
		return math.NaN()
	}
}

// ── lifecycle ───────────────────────────────────────────────────────────────

// Reload rebuilds the engine from the store. Safe to call repeatedly (after any
// graph mutation). It is a no-op-safe full restart of the running generation.
func (e *Engine) Reload() {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Tear down the previous generation and wait for its goroutines to exit.
	if e.cancel != nil {
		e.cancel()
		e.cancel = nil
	}
	if e.wg != nil {
		e.wg.Wait()
		e.wg = nil
	}

	e.liveMu.Lock()
	e.live = map[string]float64{}
	e.liveMu.Unlock()

	graphs := e.store.List()
	var compiled []*compiledGraph
	refs := map[string]RefLite{}
	for i := range graphs {
		g := graphs[i]
		if !g.Enabled {
			continue
		}
		cg := compile(g)
		compiled = append(compiled, cg)
		for k, r := range cg.refs {
			refs[k] = r
		}
	}
	if len(compiled) == 0 {
		return
	}

	genCtx, cancel := context.WithCancel(e.root)
	wg := &sync.WaitGroup{}
	e.cancel = cancel
	e.wg = wg

	for _, cg := range compiled {
		cg.engine = e
		cg.genCtx = genCtx
		cg.wg = wg
		e.registerFire(cg.id, cg.fireCh)
	}

	// One shared updates channel feeds a single dispatch goroutine; every
	// subscription delivers into it. Subscriptions are released on teardown.
	updates := make(chan broker.Update, 128)
	var unsubs []func()
	for _, r := range refs {
		unsub, err := e.broker.Subscribe(broker.SignalRef{DS: r.DS, Name: r.Name}, updates)
		if err != nil {
			e.log.Warn("control logic: subscribe failed", "ds", r.DS, "signal", r.Name, "err", err)
			continue
		}
		unsubs = append(unsubs, unsub)
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		<-genCtx.Done()
		for _, u := range unsubs {
			u()
		}
		for _, cg := range compiled {
			e.unregisterFire(cg.id, cg.fireCh)
		}
	}()

	// Dispatch goroutine: keep the live cache fresh and drive level/edge triggers.
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case u := <-updates:
				val := toNum(u.Value.Data)
				key := refKey(u.Ref.DS, u.Ref.Name)
				e.liveMu.Lock()
				e.live[key] = val
				e.liveMu.Unlock()
				for _, cg := range compiled {
					cg.onSignal(key, val)
				}
			case <-genCtx.Done():
				return
			}
		}
	}()

	// Start timer and cron triggers.
	for _, cg := range compiled {
		cg.startTriggers()
	}

	e.log.Info("control logic engine reloaded", "graphs", len(compiled), "signals", len(refs))
}

// liveGet reads the current value of a signal from the shared cache.
func (e *Engine) liveGet(ds, name string) float64 {
	if ds == "sys" {
		if name == "time" {
			return float64(time.Now().UnixNano()) / 1e9
		}
		return math.NaN() // sys:dt handled per-activation
	}
	e.liveMu.RLock()
	defer e.liveMu.RUnlock()
	v, ok := e.live[refKey(ds, name)]
	if !ok {
		return math.NaN()
	}
	return v
}

// write applies an action.write/lua-set to a target: a bare name updates a
// graph-local var; a ds:name target writes to the data source.
func (e *Engine) write(cg *compiledGraph, target string, val float64) {
	ds, name, ok := parseRef(target)
	if !ok || math.IsNaN(val) {
		return
	}
	if ds == "local" {
		cg.setLocal(name, val)
		return
	}
	if cg.dryRun {
		return // simulate: no real data-source write
	}
	src, ok := e.broker.Source(ds)
	if !ok {
		e.log.Warn("control logic: write to unknown data source", "ds", ds, "signal", name)
		return
	}
	ev := audit.Event{
		Actor:     cg.name,
		ActorType: audit.ActorSystem,
		Action:    "signal.write",
		DS:        ds,
		Signal:    name,
		Value:     strconv.FormatFloat(val, 'g', -1, 64),
		Detail:    "control logic: " + cg.name,
		Outcome:   audit.OutcomeOK,
	}
	if err := src.Write(e.root, name, val); err != nil {
		e.log.Warn("control logic: write failed", "ds", ds, "signal", name, "err", err)
		ev.Outcome = audit.OutcomeError
		ev.Error = err.Error()
	}
	e.audit.Record(ev)
}

// applyConfig resolves a config instance + its set and writes every parameter
// to its target signal via the owning data source (confmgr.Apply). Each write is
// audited. A bare instance id or a missing config store is a no-op (logged).
func (e *Engine) applyConfig(cg *compiledGraph, instanceID string) {
	if e.cfg == nil || instanceID == "" {
		return
	}
	inst, err := e.cfg.GetInstance(instanceID)
	if err != nil {
		e.log.Warn("control logic: config apply: unknown instance", "instance", instanceID, "err", err)
		return
	}
	set, err := e.cfg.SetForInstance(inst)
	if err != nil {
		e.log.Warn("control logic: config apply: set lookup failed", "instance", instanceID, "err", err)
		return
	}
	write := func(ds, signal string, value any) error {
		ev := audit.Event{
			Actor:     cg.name,
			ActorType: audit.ActorSystem,
			Action:    "signal.write",
			DS:        ds,
			Signal:    signal,
			Value:     formatAny(value),
			Detail:    "control logic config apply: " + inst.Name,
			Outcome:   audit.OutcomeOK,
		}
		var werr error
		if ds == "local" {
			cg.setLocal(signal, toNum(value))
		} else if src, ok := e.broker.Source(ds); ok {
			werr = src.Write(e.root, signal, value)
		} else {
			werr = errUnknownSource
		}
		if werr != nil {
			ev.Outcome = audit.OutcomeError
			ev.Error = werr.Error()
		}
		e.audit.Record(ev)
		return werr
	}
	confmgr.Apply(set, inst, write)
}

// readConfigParam resolves a single parameter's value from a config instance,
// coercing it to float64 for use as a control-logic value. Returns ok=false if
// the store/instance/param is missing or the value isn't numeric.
func (e *Engine) readConfigParam(instanceID, key string) (float64, bool) {
	if e.cfg == nil || instanceID == "" || key == "" {
		return 0, false
	}
	inst, err := e.cfg.GetInstance(instanceID)
	if err != nil {
		e.log.Warn("control logic: config read: unknown instance", "instance", instanceID, "err", err)
		return 0, false
	}
	set, err := e.cfg.SetForInstance(inst)
	if err != nil {
		e.log.Warn("control logic: config read: set lookup failed", "instance", instanceID, "err", err)
		return 0, false
	}
	for _, p := range set.Parameters {
		if p.Key != key {
			continue
		}
		v, ok := inst.Resolve(p)
		if !ok {
			return 0, false
		}
		f := toNum(v)
		if math.IsNaN(f) {
			return 0, false
		}
		return f, true
	}
	return 0, false
}

// writeConfigParam sets a single parameter's value on a config instance and
// saves a new revision (git-style). The value is coerced to the parameter's
// declared type (numeric/bool/string) so it validates against the set. A
// missing store/instance/param is a no-op (logged). Audited.
func (e *Engine) writeConfigParam(cg *compiledGraph, instanceID, key string, val float64) {
	if e.cfg == nil || instanceID == "" || key == "" {
		return
	}
	inst, err := e.cfg.GetInstance(instanceID)
	if err != nil {
		e.log.Warn("control logic: config write: unknown instance", "instance", instanceID, "err", err)
		return
	}
	set, err := e.cfg.SetForInstance(inst)
	if err != nil {
		e.log.Warn("control logic: config write: set lookup failed", "instance", instanceID, "err", err)
		return
	}
	var param *confmgr.Parameter
	for i := range set.Parameters {
		if set.Parameters[i].Key == key {
			param = &set.Parameters[i]
			break
		}
	}
	if param == nil {
		e.log.Warn("control logic: config write: unknown parameter", "instance", instanceID, "key", key)
		return
	}
	if inst.Values == nil {
		inst.Values = map[string]any{}
	}
	inst.Values[key] = coerceParamValue(*param, val)
	ev := audit.Event{
		Actor:     cg.name,
		ActorType: audit.ActorSystem,
		Action:    "config.instance.update",
		Detail:    "control logic config write: " + inst.Name + " " + key + "=" + formatAny(inst.Values[key]),
		Outcome:   audit.OutcomeOK,
	}
	if _, err := e.cfg.UpdateInstance(instanceID, inst, ""); err != nil {
		e.log.Warn("control logic: config write failed", "instance", instanceID, "key", key, "err", err)
		ev.Outcome = audit.OutcomeError
		ev.Error = err.Error()
	}
	e.audit.Record(ev)
}

// createConfigInstance creates a new config instance for setID, optionally
// copying values from an existing instance (fromID). A missing store/set is a
// no-op (logged). Audited. The new instance's id is logged but not written back
// to a flow variable (control-logic variables are numeric, ids are strings).
func (e *Engine) createConfigInstance(cg *compiledGraph, setID, name, fromID string) {
	if e.cfg == nil || setID == "" {
		return
	}
	if name == "" {
		name = "auto"
	}
	values := map[string]any{}
	if fromID != "" {
		if src, err := e.cfg.GetInstance(fromID); err == nil {
			for k, v := range src.Values {
				values[k] = v
			}
		} else {
			e.log.Warn("control logic: config create: copy source missing", "from", fromID, "err", err)
		}
	}
	inst := confmgr.ConfigInstance{Name: name, SetID: setID, Values: values}
	ev := audit.Event{
		Actor:     cg.name,
		ActorType: audit.ActorSystem,
		Action:    "config.instance.create",
		Detail:    "control logic config create: set=" + setID + " name=" + name,
		Outcome:   audit.OutcomeOK,
	}
	out, err := e.cfg.CreateInstance(inst, "")
	if err != nil {
		e.log.Warn("control logic: config create failed", "set", setID, "err", err)
		ev.Outcome = audit.OutcomeError
		ev.Error = err.Error()
	} else {
		ev.Detail += " -> " + out.ID
	}
	e.audit.Record(ev)
}

// snapshotConfig captures the current value of every target signal of a set and
// stores them as a new config instance. A missing store/set is a no-op (logged).
// Audited. The new instance's id is logged but not written back to a flow
// variable (control-logic variables are numeric, ids are strings).
func (e *Engine) snapshotConfig(cg *compiledGraph, setID, name string) {
	if e.cfg == nil || setID == "" {
		return
	}
	set, err := e.cfg.GetSet(setID)
	if err != nil {
		e.log.Warn("control logic: config snapshot: unknown set", "set", setID, "err", err)
		return
	}
	ctx, cancel := context.WithTimeout(e.root, 5*time.Second)
	defer cancel()
	read := func(ds, signal string) (any, error) {
		if ds == "local" {
			return cg.getLocal(signal), nil
		}
		v, err := e.broker.ReadNow(ctx, broker.SignalRef{DS: ds, Name: signal})
		if err != nil {
			return nil, err
		}
		return v.Data, nil
	}
	snap := confmgr.Snapshot(set, read)
	if name == "" {
		name = set.Name + " snapshot"
	}
	inst := confmgr.ConfigInstance{Name: name, SetID: setID, Values: snap.Values}
	ev := audit.Event{
		Actor:     cg.name,
		ActorType: audit.ActorSystem,
		Action:    "config.instance.snapshot",
		Detail:    "control logic config snapshot: set=" + setID + " name=" + name,
		Outcome:   audit.OutcomeOK,
	}
	out, err := e.cfg.CreateInstance(inst, "")
	if err != nil {
		e.log.Warn("control logic: config snapshot failed", "set", setID, "err", err)
		ev.Outcome = audit.OutcomeError
		ev.Error = err.Error()
	} else {
		ev.Detail += " -> " + out.ID + " captured=" + strconv.Itoa(snap.Captured) + " failed=" + strconv.Itoa(snap.Failed)
	}
	e.audit.Record(ev)
}

// coerceParamValue converts a numeric flow value to the Go type a config
// parameter expects, so the resulting instance validates against its set.
func coerceParamValue(p confmgr.Parameter, val float64) any {
	switch p.Type {
	case confmgr.TypeInt:
		return int64(val)
	case confmgr.TypeBool:
		return val != 0
	case confmgr.TypeString, confmgr.TypeEnum:
		return strconv.FormatFloat(val, 'g', -1, 64)
	default:
		return val
	}
}

// emitDialog delivers an action.dialog node's request to the installed Notifier
// (the WebSocket dialog hub). It is lock-free so it never deadlocks against a
// concurrent Reload that is waiting for this flow goroutine to finish.
func (e *Engine) emitDialog(n Node) {
	box, _ := e.notifier.Load().(notifierBox)
	if box.n == nil {
		return
	}
	kind := strings.TrimSpace(n.param("kind"))
	if kind != "error" && kind != "input" {
		kind = "info"
	}
	box.n.Notify(Dialog{
		ID:      strconv.FormatUint(atomic.AddUint64(&dialogSeq, 1), 10),
		Kind:    kind,
		Title:   n.param("title"),
		Message: n.param("message"),
		Target:  strings.TrimSpace(n.param("target")),
		Users:   splitCSV(n.param("users")),
		Groups:  splitCSV(n.param("groups")),
	})
}

// ── compiled graph ─────────────────────────────────────────────────────────────

type wireOut struct {
	to   string
	port string
}

// compiledGraph holds the runtime state for one enabled graph.
type compiledGraph struct {
	engine *Engine
	genCtx context.Context
	wg     *sync.WaitGroup

	// dryRun suppresses real side effects (data-source writes, config mutations,
	// dialogs) — used by the debug "simulate" sandbox. Local-variable writes stay
	// live so the flow still computes correctly. alwaysDebug forces emitDebug to
	// fire regardless of debugWatch (the sandbox is its own subscriber).
	dryRun      bool
	alwaysDebug bool

	id   string
	name string
	byId map[string]Node
	out  map[string][]wireOut
	inc  map[string][]string // incoming source ids per node (for gates)
	refs map[string]RefLite  // unique signals to subscribe (excl. sys/local)

	watchers map[string][]string // signal key → trigger node ids
	luaNodes map[string]*luaRuntime

	// fireCh receives node ids of triggers the debug UI wants to fire manually.
	// Drained by a generation goroutine (started in startTriggers) so the wg.Add
	// in activate always happens on a tracked goroutine.
	fireCh chan string

	stateMu    sync.Mutex
	levelState map[string]bool    // current truth of level triggers (threshold/alarm)
	prevBool   map[string]bool    // edge detection for threshold/alarm
	prevVal    map[string]float64 // last value for change triggers
	hasVal     map[string]bool
	lastFire   map[string]int64 // ns wall clock each trigger last fired
	locals     map[string]float64
}

func compile(g Graph) *compiledGraph {
	cg := &compiledGraph{
		id:         g.ID,
		name:       g.Name,
		byId:       map[string]Node{},
		out:        map[string][]wireOut{},
		inc:        map[string][]string{},
		refs:       map[string]RefLite{},
		watchers:   map[string][]string{},
		luaNodes:   map[string]*luaRuntime{},
		levelState: map[string]bool{},
		prevBool:   map[string]bool{},
		prevVal:    map[string]float64{},
		hasVal:     map[string]bool{},
		lastFire:   map[string]int64{},
		locals:     map[string]float64{},
		fireCh:     make(chan string, 16),
	}
	for _, n := range g.Nodes {
		cg.byId[n.ID] = n
	}
	for _, w := range g.Wires {
		port := w.FromPort
		if port == "" {
			port = "out"
		}
		cg.out[w.From] = append(cg.out[w.From], wireOut{to: w.To, port: port})
		cg.inc[w.To] = append(cg.inc[w.To], w.From)
	}

	want := func(ds, name string) {
		if ds == "sys" || ds == "local" {
			return
		}
		if name == "" {
			return
		}
		cg.refs[refKey(ds, name)] = RefLite{DS: ds, Name: name}
	}
	wantExpr := func(expr string) {
		for _, r := range CollectRefs(expr) {
			want(r.DS, r.Name)
		}
	}

	for _, n := range g.Nodes {
		switch n.Kind {
		case "trigger.threshold", "trigger.change", "trigger.alarm":
			if ds, name, ok := parseRef(n.param("signal")); ok {
				want(ds, name)
				key := refKey(ds, name)
				cg.watchers[key] = append(cg.watchers[key], n.ID)
			}
		case "flow.if":
			wantExpr(n.param("cond"))
		case "flow.loop":
			if n.param("mode") == "while" {
				wantExpr(n.param("cond"))
			}
		case "action.write", "action.log":
			wantExpr(n.param("expr"))
		case "action.lua":
			cg.luaNodes[n.ID] = newLuaRuntime(n.param("script"))
			for _, r := range luaGetRefs(n.param("script")) {
				want(r.DS, r.Name)
			}
		}
	}
	return cg
}

func (cg *compiledGraph) setLocal(name string, v float64) {
	cg.stateMu.Lock()
	cg.locals[name] = v
	cg.stateMu.Unlock()
}

func (cg *compiledGraph) getLocal(name string) float64 {
	cg.stateMu.Lock()
	defer cg.stateMu.Unlock()
	v, ok := cg.locals[name]
	if !ok {
		return 0
	}
	return v
}

// startTriggers launches timer and cron trigger goroutines for the generation.
func (cg *compiledGraph) startTriggers() {
	// Manual-fire listener: drain fireCh on a tracked goroutine so activate's
	// wg.Add never races the generation teardown's wg.Wait. Only fires nodes that
	// are actual triggers in this graph.
	cg.wg.Add(1)
	go func() {
		defer cg.wg.Done()
		for {
			select {
			case id := <-cg.fireCh:
				if n, ok := cg.byId[id]; ok && strings.HasPrefix(n.Kind, "trigger.") {
					cg.activate(id)
				}
			case <-cg.genCtx.Done():
				return
			}
		}
	}()

	hasCron := false
	for _, n := range cg.byId {
		switch n.Kind {
		case "trigger.timer":
			node := n
			cg.wg.Add(1)
			go func() {
				defer cg.wg.Done()
				d := intervalOf(node)
				t := time.NewTicker(d)
				defer t.Stop()
				for {
					select {
					case <-t.C:
						cg.activate(node.ID)
					case <-cg.genCtx.Done():
						return
					}
				}
			}()
		case "trigger.cron":
			hasCron = true
		}
	}
	if hasCron {
		cg.startCron()
	}
}

func (cg *compiledGraph) startCron() {
	type cronNode struct {
		id    string
		sched *Schedule
	}
	var crons []cronNode
	for _, n := range cg.byId {
		if n.Kind != "trigger.cron" {
			continue
		}
		sched, err := ParseSchedule(n.param("spec"))
		if err != nil {
			cg.engine.log.Warn("control logic: bad cron spec", "graph", cg.name, "spec", n.param("spec"), "err", err)
			continue
		}
		crons = append(crons, cronNode{id: n.ID, sched: sched})
	}
	if len(crons) == 0 {
		return
	}
	cg.wg.Add(1)
	go func() {
		defer cg.wg.Done()
		t := time.NewTicker(time.Second)
		defer t.Stop()
		lastMinute := -1
		for {
			select {
			case now := <-t.C:
				minute := now.Hour()*60 + now.Minute()
				if minute == lastMinute {
					continue // fire at most once per minute
				}
				lastMinute = minute
				for _, c := range crons {
					if c.sched.Match(now) {
						cg.activate(c.id)
					}
				}
			case <-cg.genCtx.Done():
				return
			}
		}
	}()
}

func intervalOf(n Node) time.Duration {
	ms, err := strconv.Atoi(strings.TrimSpace(n.param("interval")))
	if err != nil || ms < 50 {
		ms = 1000
	}
	return time.Duration(ms) * time.Millisecond
}

// ── trigger evaluation ─────────────────────────────────────────────────────────

// onSignal drives threshold/alarm (rising edge) and change triggers when a
// watched signal updates.
func (cg *compiledGraph) onSignal(key string, value float64) {
	for _, id := range cg.watchers[key] {
		node, ok := cg.byId[id]
		if !ok {
			continue
		}
		switch node.Kind {
		case "trigger.threshold":
			cur := testThreshold(value, node.param("op"), parseFloat(node.param("value")))
			cg.stateMu.Lock()
			cg.levelState[id] = cur
			prev := cg.prevBool[id]
			cg.prevBool[id] = cur
			cg.stateMu.Unlock()
			if cur && !prev {
				cg.activate(id)
			}
		case "trigger.alarm":
			lo := parseFloat(node.param("min"))
			hi := parseFloat(node.param("max"))
			cur := !math.IsNaN(value) && (value < lo || value > hi)
			cg.stateMu.Lock()
			cg.levelState[id] = cur
			prev := cg.prevBool[id]
			cg.prevBool[id] = cur
			cg.stateMu.Unlock()
			if cur && !prev {
				cg.activate(id)
			}
		case "trigger.change":
			cg.stateMu.Lock()
			had := cg.hasVal[id]
			prev := cg.prevVal[id]
			cg.prevVal[id] = value
			cg.hasVal[id] = true
			cg.stateMu.Unlock()
			if had && value != prev {
				cg.activate(id)
			}
		}
	}
}

func testThreshold(val float64, op string, cmp float64) bool {
	if math.IsNaN(val) {
		return false
	}
	switch op {
	case "<":
		return val < cmp
	case ">=":
		return val >= cmp
	case "<=":
		return val <= cmp
	case "==":
		return val == cmp
	case "!=":
		return val != cmp
	default: // ">"
		return val > cmp
	}
}

func parseFloat(s string) float64 {
	f, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
	if err != nil {
		return 0
	}
	return f
}

// ── execution ──────────────────────────────────────────────────────────────────

type runCtx struct {
	fired   string
	steps   int
	resolve Resolver
}

// activate spawns a flow run for a trigger on its own goroutine so that
// action.delay does not block signal dispatch or other flows.
func (cg *compiledGraph) activate(triggerID string) {
	now := time.Now().UnixNano()
	cg.stateMu.Lock()
	last, had := cg.lastFire[triggerID]
	cg.lastFire[triggerID] = now
	cg.stateMu.Unlock()
	dt := 0.0
	if had {
		dt = float64(now-last) / 1e9
	}

	resolve := func(ds, name string) Value { // shim: was float64 (Task 4 removes)
		switch ds {
		case "sys":
			if name == "dt" {
				return dt
			}
			return cg.engine.liveGet("sys", name)
		case "local":
			return cg.getLocal(name)
		default:
			return cg.engine.liveGet(ds, name)
		}
	}

	cg.emitDebug(triggerID, 0, false)

	cg.wg.Add(1)
	go func() {
		defer cg.wg.Done()
		ctx := &runCtx{fired: triggerID, resolve: resolve}
		cg.follow(triggerID, "out", ctx)
	}()
}

func (cg *compiledGraph) follow(fromID, port string, ctx *runCtx) {
	for _, w := range cg.out[fromID] {
		if w.port == port {
			cg.run(w.to, ctx)
		}
	}
}

func (cg *compiledGraph) run(nodeID string, ctx *runCtx) {
	if ctx.steps > maxSteps {
		return
	}
	ctx.steps++
	node, ok := cg.byId[nodeID]
	if !ok {
		return
	}

	select {
	case <-cg.genCtx.Done():
		return
	default:
	}

	cg.emitDebug(node.ID, 0, false)

	switch node.Kind {
	case "gate.and":
		if cg.gateSatisfied(node.ID, ctx.fired) {
			cg.follow(node.ID, "out", ctx)
		}

	case "flow.if":
		branch := "else"
		pass := EvalBool(node.param("cond"), ctx.resolve)
		if pass {
			branch = "then"
		}
		v := 0.0
		if pass {
			v = 1
		}
		cg.emitDebug(node.ID, v, true)
		cg.follow(node.ID, branch, ctx)

	case "flow.loop":
		if node.param("mode") == "while" {
			for i := 0; i < maxLoop && ctx.steps <= maxSteps && EvalBool(node.param("cond"), ctx.resolve); i++ {
				cg.follow(node.ID, "body", ctx)
			}
		} else {
			n := int(EvalExpr(node.param("count"), ctx.resolve))
			if n < 0 {
				n = 0
			}
			if n > maxLoop {
				n = maxLoop
			}
			for i := 0; i < n && ctx.steps <= maxSteps; i++ {
				cg.follow(node.ID, "body", ctx)
			}
		}
		cg.follow(node.ID, "done", ctx)

	case "action.write":
		val := EvalExpr(node.param("expr"), ctx.resolve)
		cg.emitDebug(node.ID, val, true)
		cg.engine.write(cg, node.param("target"), val)
		cg.follow(node.ID, "out", ctx)

	case "action.config.apply":
		if !cg.dryRun {
			cg.engine.applyConfig(cg, strings.TrimSpace(node.param("instance")))
		}
		cg.follow(node.ID, "out", ctx)

	case "action.config.read":
		v, ok := cg.engine.readConfigParam(strings.TrimSpace(node.param("instance")), strings.TrimSpace(node.param("key")))
		if ok {
			cg.engine.write(cg, node.param("target"), v)
		}
		cg.follow(node.ID, "out", ctx)

	case "action.config.write":
		val := EvalExpr(node.param("expr"), ctx.resolve)
		cg.emitDebug(node.ID, val, true)
		if !cg.dryRun {
			cg.engine.writeConfigParam(cg, strings.TrimSpace(node.param("instance")), strings.TrimSpace(node.param("key")), val)
		}
		cg.follow(node.ID, "out", ctx)

	case "action.config.create":
		if !cg.dryRun {
			cg.engine.createConfigInstance(cg, strings.TrimSpace(node.param("set")), strings.TrimSpace(node.param("name")), strings.TrimSpace(node.param("from")))
		}
		cg.follow(node.ID, "out", ctx)

	case "action.config.snapshot":
		if !cg.dryRun {
			cg.engine.snapshotConfig(cg, strings.TrimSpace(node.param("set")), strings.TrimSpace(node.param("name")))
		}
		cg.follow(node.ID, "out", ctx)

	case "action.delay":
		ms := 0
		if v, err := strconv.Atoi(strings.TrimSpace(node.param("ms"))); err == nil && v > 0 {
			ms = v
		}
		if ms > 0 {
			t := time.NewTimer(time.Duration(ms) * time.Millisecond)
			select {
			case <-t.C:
			case <-cg.genCtx.Done():
				t.Stop()
				return
			}
		}
		cg.follow(node.ID, "out", ctx)

	case "action.log":
		val := EvalExpr(node.param("expr"), ctx.resolve)
		cg.emitDebug(node.ID, val, true)
		label := strings.TrimSpace(node.param("label"))
		cg.engine.log.Info("control logic log", "graph", cg.name, "label", label, "value", val)
		cg.follow(node.ID, "out", ctx)

	case "action.lua":
		cg.runLua(node.ID, ctx)
		cg.follow(node.ID, "out", ctx)

	case "action.dialog":
		if !cg.dryRun {
			cg.engine.emitDialog(node)
		}
		cg.follow(node.ID, "out", ctx)

	default:
		cg.follow(node.ID, "out", ctx)
	}
}

// gateSatisfied: every incoming trigger must currently be satisfied. The firing
// trigger counts as satisfied; level triggers (threshold/alarm) use their truth.
func (cg *compiledGraph) gateSatisfied(gateID, fired string) bool {
	inputs := cg.inc[gateID]
	if len(inputs) == 0 {
		return false
	}
	cg.stateMu.Lock()
	defer cg.stateMu.Unlock()
	for _, src := range inputs {
		if src == fired || cg.levelState[src] {
			continue
		}
		return false
	}
	return true
}

// luaGetRefs scans a Lua script for get("ds:name") / get('ds:name') literals so
// the engine can subscribe to the signals the script reads.
var luaGetRe = regexp.MustCompile(`get\s*\(\s*["']([^"']+)["']`)

func luaGetRefs(script string) []RefLite {
	var out []RefLite
	for _, m := range luaGetRe.FindAllStringSubmatch(script, -1) {
		if ds, name, ok := parseRef(m[1]); ok {
			out = append(out, RefLite{DS: ds, Name: name})
		}
	}
	return out
}

func (cg *compiledGraph) runLua(nodeID string, ctx *runCtx) {
	lr := cg.luaNodes[nodeID]
	if lr == nil {
		return
	}
	lr.run(ctx.resolve, func(target string, val float64) {
		cg.engine.write(cg, target, val)
	}, func(msg string) {
		cg.engine.log.Info("control logic lua", "graph", cg.name, "msg", msg)
	})
}