package automations import ( "context" "encoding/json" "fmt" "log/slog" "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" ) // Action is a unit of work executed when a schedule, event rule or reminder // fires. Execute runs the action with the trigger payload and returns a result // (or an error, wrapped with %w). Implementations must be safe for concurrent use // because the engine may invoke them from multiple goroutines. // // The three concrete implementations are FuncAction (a plain function), ToolAction // (an sdk.Tool) and AIAction (a prompt run through the AI pipeline). type Action interface { // Name identifies the action in logs and in the published Result. Name() string // Execute runs the action with the trigger payload. Execute(ctx context.Context, payload any) (any, error) } // Func is a plain function action. type Func func(ctx context.Context, payload any) (any, error) // FuncAction adapts a plain function to the Action interface. type FuncAction struct { name string fn Func } // NewFuncAction wraps fn as a named action. It returns an error wrapping // ErrNilAction when name is blank or fn is nil. func NewFuncAction(name string, fn Func) (*FuncAction, error) { if strings.TrimSpace(name) == "" { return nil, fmt.Errorf("automations: func action: %w", ErrNilAction) } if fn == nil { return nil, fmt.Errorf("automations: func action %q: %w", name, ErrNilAction) } return &FuncAction{name: name, fn: fn}, nil } // Name returns the action name. func (f *FuncAction) Name() string { return f.name } // Execute runs the wrapped function with the payload. func (f *FuncAction) Execute(ctx context.Context, payload any) (any, error) { return f.fn(ctx, payload) } // ToolAction adapts an sdk.Tool to the Action interface. type ToolAction struct { Tool sdk.Tool } // NewToolAction wraps a registered tool as a named action. A nil tool returns an // error wrapping ErrNilAction. func NewToolAction(t sdk.Tool) (*ToolAction, error) { if t == nil { return nil, fmt.Errorf("automations: tool action: %w", ErrNilAction) } return &ToolAction{Tool: t}, nil } // Name returns the underlying tool's name. func (t *ToolAction) Name() string { return t.Tool.Name() } // Execute delegates to the underlying tool. func (t *ToolAction) Execute(ctx context.Context, payload any) (any, error) { return t.Tool.Execute(ctx, payload) } // AIAction drives a prompt through the AI pipeline (loop or agent) and returns // the generated text. sessionID scopes the conversation for tracing/persistence. type AIAction struct { name string prompt string sessionID string pipeline Pipeline } // NewAIAction builds a prompting action. name must be non-blank and p must be // non-nil; otherwise an error wrapping ErrNilAction or ErrNoPipeline is returned. func NewAIAction(name, prompt, sessionID string, p Pipeline) (*AIAction, error) { if strings.TrimSpace(name) == "" { return nil, fmt.Errorf("automations: ai action: %w", ErrNilAction) } if p == nil { return nil, fmt.Errorf("automations: ai action %q: %w", name, ErrNoPipeline) } return &AIAction{name: name, prompt: prompt, sessionID: sessionID, pipeline: p}, nil } // Name returns the action name. func (a *AIAction) Name() string { return a.name } // Execute runs the prompt (with the payload appended) through the pipeline and // returns the generated text as the result. func (a *AIAction) Execute(ctx context.Context, payload any) (any, error) { msgs := []model.Message{model.TextMessage(model.RoleUser, a.promptMessage(payload))} res, err := a.pipeline.Run(ctx, a.sessionID, msgs) if err != nil { return nil, fmt.Errorf("automations: ai action %q: %w", a.name, err) } return res.Text, nil } // promptMessage renders the prompt plus a readable form of the trigger payload. func (a *AIAction) promptMessage(payload any) string { if payload == nil { return a.prompt } if s, ok := payload.(string); ok { return a.prompt + "\nTrigger payload: " + s } if b, err := json.Marshal(payload); err == nil { return a.prompt + "\nTrigger payload: " + string(b) } return a.prompt + "\nTrigger payload: " + fmt.Sprint(payload) } // runAction executes a with payload under the configured deadline and publishes // the outcome to SubjectResult. It is the single place where automation execution // and result reporting are wired, so every trigger kind reports consistently. A // panic is recovered and recorded as a failed result so a misbehaving action // cannot crash the calling goroutine. func runAction(ctx context.Context, cfg Config, b bus.Bus, log *slog.Logger, kind, ruleID string, a Action, payload any) { if a == nil { return } execCtx := ctx if cfg.ActionTimeout > 0 { var cancel context.CancelFunc execCtx, cancel = context.WithTimeout(ctx, cfg.ActionTimeout) defer cancel() } out, err := executeSafely(execCtx, a, payload) res := Result{Kind: kind, RuleID: ruleID, Action: a.Name(), Payload: payload, Output: out} if err != nil { res.Error = err.Error() log.Error("automation action failed", "kind", kind, "rule_id", ruleID, "action", a.Name(), "err", err) } else { log.Info("automation action ran", "kind", kind, "rule_id", ruleID, "action", a.Name()) } if cerr := b.Publish(ctx, SubjectResult, res); cerr != nil { log.Error("publish automation result", "kind", kind, "rule_id", ruleID, "err", cerr) } } // executeSafely runs a.Execute, converting a panic into an error so the action // reports a failed result instead of crashing the engine goroutine. func executeSafely(ctx context.Context, a Action, payload any) (out any, err error) { defer func() { if r := recover(); r != nil { out = nil err = fmt.Errorf("automations: action %q panicked: %v", a.Name(), r) } }() return a.Execute(ctx, payload) }