Backlog/v12 new soar flows

backend/database/migrations.go

@@ -50,6 +50,8 @@ func Models() []any {
 		alerts_domain.AlertTag{},
 		alerts_domain.AlertTagRule{},
 		arr_domain.SoarExecution{},
+		arr_domain.SoarFlowRun{},
+		arr_domain.SoarExecutionEdge{},
 		arr_domain.SoarVariable{},
 		compliance_domain.ReportSchedule{},
 		compliance_domain.Report{},
@@ -73,6 +75,21 @@ func MigrateDatabase(db *gorm.DB, migrationsURL string) error {
 		catcher.Warn("could not create uuid-ossp extension", map[string]any{"error": err.Error()})
 	}
 
+	// ponytail: v11 soar_executions rows predate the DAG columns. GORM
+	// AutoMigrate can't add NOT NULL columns without a default when the table
+	// already has rows, so we pre-create them here with DEFAULT ''. New inserts
+	// override the default; legacy rows keep empty strings, which is what the
+	// DAG dispatcher already tolerates for terminal history.
+	for _, stmt := range []string{
+		`ALTER TABLE IF EXISTS soar_executions ADD COLUMN IF NOT EXISTS node_id  varchar(150) NOT NULL DEFAULT ''`,
+		`ALTER TABLE IF EXISTS soar_executions ADD COLUMN IF NOT EXISTS kind     varchar(20)  NOT NULL DEFAULT ''`,
+		`ALTER TABLE IF EXISTS soar_executions ADD COLUMN IF NOT EXISTS executor varchar(60)  NOT NULL DEFAULT ''`,
+	} {
+		if err := db.Exec(stmt).Error; err != nil {
+			return err
+		}
+	}
+
 	models := Models()
 	if len(models) > 0 {
 		catcher.Info("running GORM AutoMigrate...", nil)

backend/modules.go

@@ -198,7 +198,16 @@ func initModules(db *gorm.DB, cfg *config) *modules {
 		agentClient = nil
 	}
 
-	soarMod := soar.NewModule(db, agentClient, signer, cipher, tenantLister)
+	// SOC-AI client is built here so the SOAR LLM executors can share it.
+	// socai.NewModule below reuses the same base URL/key.
+	socAIClient := socai.NewSocAIClient(cfg.socAIBaseURL, cfg.internalKey)
+	// Notifications module is built early so its usecase can back the SOAR
+	// notify executor. Its own dependencies (db + audit logger + leases) are
+	// already available at this point.
+	notificationsMod := notifications.NewModule(db, auditMod.Logger(), joblease.New(db),
+		env.Int("NOTIFICATIONS_READ_RETENTION_DAYS", 30, false),
+		env.Int("NOTIFICATIONS_RETENTION_DAYS", 365, false))
+	soarMod := soar.NewModule(db, agentClient, signer, cipher, socAIClient, notificationsMod.Producer(), tenantLister)
 	eventProcessingMod := eventprocessing.NewModule(db, events, auditMod.Logger(), cfg.playgroundBaseURL, cfg.internalKey)
 
 	alertsMod.SetCorrelationResolver(eventProcessingMod)
@@ -213,10 +222,6 @@ func initModules(db *gorm.DB, cfg *config) *modules {
 	}
 	datasourcesMod := datasources.NewModule(dsUC, dsReconciler, agentClient)
 
-	notificationsMod := notifications.NewModule(db, auditMod.Logger(), joblease.New(db),
-		env.Int("NOTIFICATIONS_READ_RETENTION_DAYS", 30, false),
-		env.Int("NOTIFICATIONS_RETENTION_DAYS", 365, false))
-
 	iam_handler.AppBaseURL = env.String("APP_BASE_URL", "", false)
 
 	// Extra purgers: ClickHouse rows and per-tenant filesystem folders.

backend/modules/mcp/tools_soar.go

@@ -21,29 +21,25 @@ func registerSOAR(m *Module) {
 // ---- soar.rule.* -----------------------------------------------------------
 
 type soarRuleCreateInput struct {
-	Name           string              `json:"name"`
-	Description    string              `json:"description,omitempty"`
-	Conditions     []dto.FilterVM      `json:"conditions"`
-	Commands       []dto.FlowCommandVM `json:"commands"`
-	Active         bool                `json:"active"`
-	AgentPlatform  string              `json:"agent_platform"`
-	DefaultAgent   string              `json:"default_agent,omitempty"`
-	Shell          string              `json:"shell,omitempty"`
-	ExcludedAgents []string            `json:"excluded_agents,omitempty"`
+	Name        string                    `json:"name"`
+	Description string                    `json:"description,omitempty"`
+	Conditions  []dto.FilterVM            `json:"conditions"`
+	Roots       []string                  `json:"roots"`
+	Nodes       map[string]dto.FlowNodeVM `json:"nodes"`
+	MaxDepth    int                       `json:"max_depth,omitempty"`
+	Active      bool                      `json:"active"`
 }
 
 type soarRuleUpdateInput struct {
-	RelPath        string              `json:"rel_path"`
-	ID             *int64              `json:"id,omitempty"`
-	Name           string              `json:"name"`
-	Description    string              `json:"description,omitempty"`
-	Conditions     []dto.FilterVM      `json:"conditions"`
-	Commands       []dto.FlowCommandVM `json:"commands"`
-	Active         bool                `json:"active"`
-	AgentPlatform  string              `json:"agent_platform"`
-	DefaultAgent   string              `json:"default_agent,omitempty"`
-	Shell          string              `json:"shell,omitempty"`
-	ExcludedAgents []string            `json:"excluded_agents,omitempty"`
+	RelPath     string                    `json:"rel_path"`
+	ID          *int64                    `json:"id,omitempty"`
+	Name        string                    `json:"name"`
+	Description string                    `json:"description,omitempty"`
+	Conditions  []dto.FilterVM            `json:"conditions"`
+	Roots       []string                  `json:"roots"`
+	Nodes       map[string]dto.FlowNodeVM `json:"nodes"`
+	MaxDepth    int                       `json:"max_depth,omitempty"`
+	Active      bool                      `json:"active"`
 }
 
 type soarRuleRelPathInput struct {
@@ -56,13 +52,12 @@ type soarRuleSetEnabledInput struct {
 }
 
 type soarRuleListInput struct {
-	RuleName      string `json:"name,omitempty"`
-	RuleActive    *bool  `json:"active,omitempty"`
-	AgentPlatform string `json:"agent_platform,omitempty"`
-	CreatedBy     string `json:"created_by,omitempty"`
-	SystemOwner   *bool  `json:"system_owner,omitempty"`
-	Page          int    `json:"page,omitempty"`
-	Size          int    `json:"size,omitempty"`
+	RuleName    string `json:"name,omitempty"`
+	RuleActive  *bool  `json:"active,omitempty"`
+	CreatedBy   string `json:"created_by,omitempty"`
+	SystemOwner *bool  `json:"system_owner,omitempty"`
+	Page        int    `json:"page,omitempty"`
+	Size        int    `json:"size,omitempty"`
 }
 
 func registerSOARRules(m *Module) {
@@ -72,15 +67,14 @@ func registerSOARRules(m *Module) {
 		Name: "soar.rule.create", Title: "Create SOAR rule",
 	}, Gate{Permission: "soar.write"},
 		func(ctx context.Context, actor *authz.Actor, in soarRuleCreateInput) (any, error) {
-			if len(in.Conditions) == 0 || len(in.Commands) == 0 {
-				return nil, fmt.Errorf("conditions and commands are required")
+			if len(in.Conditions) == 0 || len(in.Roots) == 0 || len(in.Nodes) == 0 {
+				return nil, fmt.Errorf("conditions, roots, and nodes are required")
 			}
 			active := in.Active
 			return uc.Create(ctx, dto.CreateRuleRequest{
 				Name: in.Name, Description: in.Description,
-				Conditions: in.Conditions, Commands: in.Commands, Active: &active,
-				AgentPlatform: in.AgentPlatform, DefaultAgent: in.DefaultAgent,
-				Shell: in.Shell, ExcludedAgents: in.ExcludedAgents,
+				Conditions: in.Conditions, Roots: in.Roots, Nodes: in.Nodes,
+				MaxDepth: in.MaxDepth, Active: &active,
 			}, actor.Email)
 		})
 
@@ -91,9 +85,8 @@ func registerSOARRules(m *Module) {
 			active := in.Active
 			return uc.Update(ctx, in.RelPath, dto.UpdateRuleRequest{
 				ID: in.ID, Name: in.Name, Description: in.Description,
-				Conditions: in.Conditions, Commands: in.Commands, Active: &active,
-				AgentPlatform: in.AgentPlatform, DefaultAgent: in.DefaultAgent,
-				Shell: in.Shell, ExcludedAgents: in.ExcludedAgents,
+				Conditions: in.Conditions, Roots: in.Roots, Nodes: in.Nodes,
+				MaxDepth: in.MaxDepth, Active: &active,
 			}, actor.Email)
 		})
 
@@ -133,7 +126,7 @@ func registerSOARRules(m *Module) {
 		func(ctx context.Context, _ *authz.Actor, in soarRuleListInput) (any, error) {
 			f := dto.RuleFilters{
 				RuleName: in.RuleName, RuleActive: in.RuleActive,
-				AgentPlatform: in.AgentPlatform, CreatedBy: in.CreatedBy, SystemOwner: in.SystemOwner,
+				CreatedBy: in.CreatedBy, SystemOwner: in.SystemOwner,
 				Params: database.Params{Page: in.Page, Size: clampPageSize(in.Size)},
 			}
 			return uc.List(ctx, f)

backend/modules/soar/connectors/repository.go

@@ -35,8 +35,62 @@ type ExecutionStatusUpdate struct {
 type ExecutionRepository interface {
 	Create(ctx context.Context, e *domain.SoarExecution) (*domain.SoarExecution, error)
 	List(ctx context.Context, f ExecutionFilters) ([]domain.SoarExecution, int64, error)
+	Get(ctx context.Context, id uuid.UUID) (*domain.SoarExecution, error)
 	UpdateStatus(ctx context.Context, id uuid.UUID, u ExecutionStatusUpdate) error
 	ClaimPending(ctx context.Context, id uuid.UUID, leaseDuration time.Duration) (bool, error)
+
+	// SaveOutput persists an enrichment node's structured output; called after
+	// a successful Executor.Execute for kind=enrichment.
+	SaveOutput(ctx context.Context, id uuid.UUID, output []byte) error
+
+	// RecordEdge inserts one parent→child edge and atomically decrements the
+	// child's pending_parents counter. If the child does not yet exist for
+	// (flowRunID, childNodeID, childDepth), it is created with the given
+	// template (status=WAITING, pending_parents=incomingCount, kind/executor
+	// copied from the flow node). Returns the child's post-update state.
+	RecordEdge(ctx context.Context, req RecordEdgeRequest) (child *domain.SoarExecution, err error)
+
+	// ListFiredParents returns all fired parents of a child execution, used to
+	// build the merged context bag when the child transitions to PENDING.
+	ListFiredParents(ctx context.Context, childID uuid.UUID) ([]domain.SoarExecution, error)
+
+	// TransitionReady moves a child from WAITING to PENDING (or DEAD) once all
+	// its parents have resolved. context/params/command/shell are the
+	// interpolated values computed by the caller from ListFiredParents.
+	TransitionReady(ctx context.Context, id uuid.UUID, ready ReadyUpdate) error
+}
+
+type RecordEdgeRequest struct {
+	FlowRunID       uuid.UUID
+	TenantID        uuid.UUID
+	RulePath        string
+	AlertID         string
+	Parent          domain.SoarExecution
+	ChildNodeID     string
+	ChildDepth      int
+	ChildKind       domain.NodeKind
+	ChildExecutor   string
+	IncomingCount   int
+	Branch          domain.EdgeBranch
+	Fired           bool
+}
+
+type ReadyUpdate struct {
+	Status  domain.ExecutionStatus
+	Context []byte
+	Params  []byte
+	Command string
+	Shell   string
+	Agent   string
+}
+
+// FlowRunRepository holds the top-level state of one root-invocation.
+type FlowRunRepository interface {
+	Create(ctx context.Context, r *domain.SoarFlowRun) (*domain.SoarFlowRun, error)
+	Get(ctx context.Context, id uuid.UUID) (*domain.SoarFlowRun, error)
+	// MaybeComplete transitions the run to COMPLETED/FAILED when no non-terminal
+	// executions remain. Returns true when a transition happened.
+	MaybeComplete(ctx context.Context, id uuid.UUID) (bool, error)
 }
 
 type ResolveFilterRepository interface {

backend/modules/soar/domain/execution.go

@@ -1,6 +1,7 @@
 package domain
 
 import (
+	"encoding/json"
 	"time"
 
 	"github.com/google/uuid"
@@ -9,16 +10,30 @@ import (
 type ExecutionStatus string
 
 const (
-	ExecutionStatusExecuted ExecutionStatus = "EXECUTED"
-	ExecutionStatusPending  ExecutionStatus = "PENDING"
-	ExecutionStatusFailed   ExecutionStatus = "FAILED"
+	ExecutionStatusWaiting   ExecutionStatus = "WAITING"
+	ExecutionStatusPending   ExecutionStatus = "PENDING"
+	ExecutionStatusExecuting ExecutionStatus = "EXECUTING"
+	ExecutionStatusExecuted  ExecutionStatus = "EXECUTED"
+	ExecutionStatusFailed    ExecutionStatus = "FAILED"
+	ExecutionStatusDead      ExecutionStatus = "DEAD"
 )
 
+// Terminal returns true if the status represents a finished execution — the
+// dispatcher will not touch it again.
+func (s ExecutionStatus) Terminal() bool {
+	switch s {
+	case ExecutionStatusExecuted, ExecutionStatusFailed, ExecutionStatusDead:
+		return true
+	}
+	return false
+}
+
 type NonExecutionCause string
 
 const (
 	NonExecutionCauseAgentOffline  NonExecutionCause = "AGENT_OFFLINE"
 	NonExecutionCauseAgentNotFound NonExecutionCause = "AGENT_NOT_FOUND"
+	NonExecutionCauseMaxDepth      NonExecutionCause = "MAX_DEPTH_EXCEEDED"
 	NonExecutionCauseUnknown       NonExecutionCause = "UNKNOWN"
 )
 
@@ -29,22 +44,85 @@ const (
 	ExecutionOriginManual ExecutionOrigin = "MANUAL"
 )
 
+type EdgeBranch string
+
+const (
+	EdgeBranchSuccess EdgeBranch = "SUCCESS"
+	EdgeBranchError   EdgeBranch = "ERROR"
+)
+
+// SoarExecution is one node instance of a running flow. A single flow run may
+// hold many rows; siblings that AND-join land on the same (flow_run_id, node_id,
+// depth) tuple and coalesce.
 type SoarExecution struct {
-	ID                uuid.UUID          `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()"                                 json:"id"`
-	TenantID          uuid.UUID          `gorm:"column:tenant_id;type:uuid;not null;index:idx_soar_execution_tenant_started,priority:1"   json:"-"`
-	Origin            ExecutionOrigin    `gorm:"column:origin;size:20;not null"      json:"origin"`
-	RulePath          string             `gorm:"column:rule_path;size:512;not null"  json:"rulePath,omitempty"`      // flow only — the YAML file, not an FK
-	AlertID           string             `gorm:"column:alert_id;size:150;not null"   json:"alertId,omitempty"`       // flow only
-	TriggeredBy       string             `gorm:"column:triggered_by;size:150;not null" json:"triggeredBy,omitempty"` // manual only
-	Agent             string             `gorm:"column:agent;size:150;not null" json:"agent"`
-	Command           string             `gorm:"column:command;not null"        json:"command"`
-	Result            string             `gorm:"column:result"                  json:"result,omitempty"`
-	Status            ExecutionStatus    `gorm:"column:status;size:100;not null" json:"status"`
+	ID       uuid.UUID  `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()"                               json:"id"`
+	TenantID uuid.UUID  `gorm:"column:tenant_id;type:uuid;not null;index:idx_soar_execution_tenant_started,priority:1" json:"-"`
+	Origin   ExecutionOrigin `gorm:"column:origin;size:20;not null" json:"origin"`
+
+	// Manual-execution fields (untouched by the DAG engine).
+	TriggeredBy string `gorm:"column:triggered_by;size:150;not null" json:"triggeredBy,omitempty"`
+
+	// Flow-run linkage — nullable for manual executions.
+	FlowRunID *uuid.UUID `gorm:"column:flow_run_id;type:uuid;index;index:idx_soar_execution_run_node,unique,priority:1" json:"flowRunId,omitempty"`
+	RulePath  string     `gorm:"column:rule_path;size:512;not null" json:"rulePath,omitempty"`
+	AlertID   string     `gorm:"column:alert_id;size:150;not null"  json:"alertId,omitempty"`
+
+	// DAG node identity.
+	NodeID string   `gorm:"column:node_id;size:150;not null;index:idx_soar_execution_run_node,unique,priority:2" json:"nodeId,omitempty"`
+	Depth  int      `gorm:"column:depth;not null;default:0;index:idx_soar_execution_run_node,unique,priority:3"  json:"depth"`
+	Kind   NodeKind `gorm:"column:kind;size:20;not null"                                                          json:"kind"`
+
+	// Executor + interpolated payload.
+	Executor string          `gorm:"column:executor;size:60;not null"     json:"executor"`
+	Params   json.RawMessage `gorm:"column:params;type:jsonb"             json:"params,omitempty"`
+	Output   json.RawMessage `gorm:"column:output;type:jsonb"             json:"output,omitempty"`
+	Context  json.RawMessage `gorm:"column:context;type:jsonb"            json:"context,omitempty"`
+
+	// Shell-executor legacy fields (kept for manual execs and shell nodes).
+	Agent   string `gorm:"column:agent;size:150;not null" json:"agent"`
+	Command string `gorm:"column:command;not null"        json:"command"`
+	Result  string `gorm:"column:result"                  json:"result,omitempty"`
+	Shell   string `gorm:"column:shell;size:20"           json:"shell,omitempty"`
+
+	// AND-join accounting.
+	PendingParents int `gorm:"column:pending_parents;not null;default:0" json:"pendingParents"`
+	DeadParents    int `gorm:"column:dead_parents;not null;default:0"    json:"deadParents"`
+
+	Status            ExecutionStatus    `gorm:"column:status;size:20;not null" json:"status"`
 	StartedAt         time.Time          `gorm:"column:started_at;not null;index:idx_soar_execution_tenant_started,priority:2,sort:desc" json:"startedAt"`
-	FinishedAt        *time.Time         `gorm:"column:finished_at"                                                                     json:"finishedAt,omitempty"`
-	ClaimedAt         *time.Time         `gorm:"column:claimed_at"                     json:"-"`
-	Retries           int                `gorm:"column:retries;not null;default:0"     json:"retries"`
-	NonExecutionCause *NonExecutionCause `gorm:"column:non_execution_cause;size:100"   json:"nonExecutionCause,omitempty"`
+	FinishedAt        *time.Time         `gorm:"column:finished_at"                                                                      json:"finishedAt,omitempty"`
+	ClaimedAt         *time.Time         `gorm:"column:claimed_at"                                                                       json:"-"`
+	Retries           int                `gorm:"column:retries;not null;default:0"                                                       json:"retries"`
+	NonExecutionCause *NonExecutionCause `gorm:"column:non_execution_cause;size:100"                                                     json:"nonExecutionCause,omitempty"`
 }
 
 func (SoarExecution) TableName() string { return "soar_executions" }
+
+// SoarFlowRun groups every node execution triggered by a single alert match.
+type SoarFlowRun struct {
+	ID         uuid.UUID       `gorm:"column:id;type:uuid;primaryKey;default:gen_random_uuid()" json:"id"`
+	TenantID   uuid.UUID       `gorm:"column:tenant_id;type:uuid;not null;index"                json:"-"`
+	RulePath   string          `gorm:"column:rule_path;size:512;not null"                       json:"rulePath"`
+	AlertID    string          `gorm:"column:alert_id;size:150;not null"                        json:"alertId"`
+	AlertJSON  json.RawMessage `gorm:"column:alert_json;type:jsonb;not null"                    json:"-"`
+	MaxDepth   int             `gorm:"column:max_depth;not null;default:50"                     json:"maxDepth"`
+	Status     ExecutionStatus `gorm:"column:status;size:20;not null"                           json:"status"`
+	StartedAt  time.Time       `gorm:"column:started_at;not null"                               json:"startedAt"`
+	FinishedAt *time.Time      `gorm:"column:finished_at"                                       json:"finishedAt,omitempty"`
+}
+
+func (SoarFlowRun) TableName() string { return "soar_flow_runs" }
+
+// SoarExecutionEdge records a resolved incoming edge on a child execution.
+// `Fired=false` means the parent branch didn't match — the edge died and the
+// child is no longer reachable through this path.
+type SoarExecutionEdge struct {
+	ChildExecID  uuid.UUID  `gorm:"column:child_exec_id;type:uuid;primaryKey"  json:"childExecId"`
+	ParentExecID uuid.UUID  `gorm:"column:parent_exec_id;type:uuid;primaryKey" json:"parentExecId"`
+	FlowRunID    uuid.UUID  `gorm:"column:flow_run_id;type:uuid;not null;index" json:"flowRunId"`
+	Branch       EdgeBranch `gorm:"column:branch;size:16;not null"             json:"branch"`
+	Fired        bool       `gorm:"column:fired;not null"                      json:"fired"`
+	CreatedAt    time.Time  `gorm:"column:created_at;not null"                 json:"createdAt"`
+}
+
+func (SoarExecutionEdge) TableName() string { return "soar_execution_edges" }