workflow

// internal/fabric/task/workflow
type NodeID string

type WorkflowSpec struct {
    ID       string
    Nodes    []NodeSpec
    Edges    []EdgeSpec
    Entries  []NodeID
    Loop     *LoopSpec
    Schedule ScheduleSpec
}

type NodeSpec struct {
    ID          NodeID
    Name        string
    AgentType   string
    Input       string
    Timeout     time.Duration
    Retry       *RetrySpec
    Recovery    *RecoverySpec
    Interrupt   *InterruptSpec
    Join        JoinKind
    Condition   *ConditionExpr
    SubWorkflow *WorkflowSpec
    Metadata    map[string]string
}

type EdgeSpec struct {
    From     NodeID
    To       NodeID
    Kind     EdgeKind
    Branch   BranchKind
    Group    string
    Priority int
    Cond     *ConditionExpr
}

type NodeStatus string  // pending, ready, running, completed, failed, interrupted, cancelled, not_selected, unreachable, blocked
type EdgeKind string    // data_dependency, control_flow
type BranchKind string  // branch_one, branch_many
type JoinKind string    // join_all, join_any, merge

type RetrySpec struct {
    MaxAttempts       int
    InitialDelay      time.Duration
    MaxDelay          time.Duration
    BackoffMultiplier float64
}

type RecoverySpec struct {
    Strategy        string // "retry", "replace_node", "fail_fast"
    ReplacementAgent string
}

type InterruptSpec struct {
    Message    string
    TimeoutSec int
    AutoAction string // "skip", "approve", "fallback"
}

type LoopSpec struct {
    MaxIterations int
    LoopNodes     []NodeID
}

type ScheduleSpec struct {
    MaxParallel int
}

type Runner struct { /* unexported */ }
type RunnerOption func(*Runner)

func NewRunner(executor NodeExecutor, opts ...RunnerOption) *Runner
func WithScheduleStrategy(strategy ScheduleStrategy) RunnerOption
func WithReadySelector(selector func([]NodeID) NodeID) RunnerOption
func WithInterruptHandler(handler func(context.Context, *InterruptSpec, StateView) (bool, error)) RunnerOption
func WithRecoveryHandler(handler func(context.Context, NodeID, error, *NodeSpec) (bool, map[string]any, error)) RunnerOption
func WithConditionEvaluator(evaluator func(*ConditionExpr, StateView) bool) RunnerOption
func WithBindings(bindings map[NodeID]func(map[string]any) bool) RunnerOption
func WithUntilCondition(predicate func(map[string]any, int) bool) RunnerOption
func WithCompiledWorkflow(compiled *CompiledWorkflow) RunnerOption
func WithInitialInput(input string) RunnerOption
func WithInitialVariables(variables map[string]string) RunnerOption
func WithInitialState(state map[string]any) RunnerOption
func WithPluginBus(bus *ares_runtime.PluginBus) RunnerOption
func WithCheckpointStore(store ares_runtime.CheckpointStore) RunnerOption
func WithExecutionID(executionID string) RunnerOption
func WithFailOnNodeError(enabled bool) RunnerOption
func WithEventSink(sink RunnerEventSink) RunnerOption
func WithPatchQueue(queue *PatchQueue) RunnerOption
func WithExecutionCollector(collector *ares_runtime.ExecutionCollector) RunnerOption

func (r *Runner) Execute(ctx context.Context, spec *WorkflowSpec) (*Result, error)
func (r *Runner) ExecuteBound(ctx context.Context, bound *BoundWorkflow) (*Result, error)
func (r *Runner) ResumeExecution(ctx context.Context, spec *WorkflowSpec, executionID string) (*Result, error)

type NodeExecutor interface {
    ExecuteNode(ctx context.Context, spec *NodeSpec, scope *ExecutionScope) (map[string]any, error)
}

type ExecutableFunc func(ctx context.Context, view StateView) (map[string]any, error)
type FuncNodeExecutor struct { /* unexported */ }
func NewFuncNodeExecutor() *FuncNodeExecutor
func (e *FuncNodeExecutor) Register(id NodeID, fn ExecutableFunc)
func (e *FuncNodeExecutor) ExecuteNode(ctx context.Context, spec *NodeSpec, scope *ExecutionScope) (map[string]any, error)

type EngineNodeExecutor struct { /* unexported */ }
func NewEngineNodeExecutor(registry *engine.AgentRegistry, steps []*engine.Step) (*EngineNodeExecutor, error)
func NewEngineNodeExecutorWithStepExecutor(/* ... */) (*EngineNodeExecutor, error)
func (e *EngineNodeExecutor) ExecuteNode(ctx context.Context, spec *NodeSpec, scope *ExecutionScope) (map[string]any, error)

type Predicate func(state map[string]any) bool
type Router func(ctx context.Context, nodeID string, state map[string]any, output string) string
type LoopPredicate func(state map[string]any, iteration int) bool

type BoundWorkflow struct { /* unexported */ }
func BindCompiledWorkflow(compiled *CompiledWorkflow) (*BoundWorkflow, error)

type CompiledWorkflow struct { /* unexported */ }
func CompileFromEngine(w *engine.Workflow) (*WorkflowSpec, error)
func CompileFromEngineWithBindings(w *engine.Workflow) (*CompiledWorkflow, error)

type ExecutionScope struct { /* unexported */ }
func NewExecutionScope(execID string, spec *WorkflowSpec) *ExecutionScope
func (s *ExecutionScope) State() StateView
func (s *ExecutionScope) Writer() StateWriter
func (s *ExecutionScope) CommitState()
func (s *ExecutionScope) StateSnapshot() map[string]any
func (s *ExecutionScope) RestoreState(state map[string]any)
func (s *ExecutionScope) NodeStatus(id NodeID) NodeStatus
func (s *ExecutionScope) SetNodeStatus(id NodeID, status NodeStatus)
func (s *ExecutionScope) SetNodeOutput(id NodeID, output map[string]any)
func (s *ExecutionScope) SetNodeError(id NodeID, err error)
func (s *ExecutionScope) ToResult() *Result

type StateView interface {
    Get(key string) (any, bool)
    GetNodeOutput(nodeID NodeID) (map[string]any, bool)
}
type StateWriter interface {
    Set(key string, value any)
    SetNodeOutput(nodeID NodeID, output map[string]any)
}

type Result struct {
    ExecutionID string
    SpecID      string
    Status      NodeStatus
    NodeStates  []*NodeStatusValue
    Duration    time.Duration
    Error       string
}

type Scheduler struct { /* unexported */ }
func NewScheduler(spec *WorkflowSpec, strategy ScheduleStrategy) (*Scheduler, error)
func (s *Scheduler) HasReady() bool
func (s *Scheduler) Next() NodeID
func (s *Scheduler) NextWithSelector(selector func([]NodeID) NodeID) (NodeID, error)
func (s *Scheduler) OnNodeCompleted(id NodeID)
func (s *Scheduler) OnNodeFailed(id NodeID)

type CheckpointSnapshot struct {
    SchemaVersion         int
    ExecutionID           string
    SpecID                string
    BaseSpecHash          string
    SpecHash              string
    EffectiveSpec         *WorkflowSpec
    State                 map[string]any
    NodeStates            []NodeStatusValue
    Scheduler             SchedulerSnapshot
    LoopIteration         int
    LoopIterationComplete bool
    LoopHistory           []LoopIteration
    PendingInterrupts     []PendingInterrupt
    MutationIDs           []string
    PendingMutations      []Mutation
    EventSequence         uint64
    SavedAt               time.Time
    CollectorData         map[string]any
}

func NewWorkflow(id string) *WorkflowSpec
func (s *WorkflowSpec) AddNode(n NodeSpec) *WorkflowSpec
func (s *WorkflowSpec) AddEdge(e EdgeSpec) *WorkflowSpec
func (s *WorkflowSpec) WithEntry(ids ...NodeID) *WorkflowSpec
func (s *WorkflowSpec) WithLoop(loop *LoopSpec) *WorkflowSpec
func (s *WorkflowSpec) WithMaxParallel(n int) *WorkflowSpec
func RunWorkflow(ctx context.Context, spec *WorkflowSpec, functions map[NodeID]ExecutableFunc, opts ...RunnerOption) (*Result, error)