executor

type Checkpoint ¶

type Checkpoint struct {
	GraphName string               `json:"graph_name"`
	RunID     string               `json:"run_id,omitempty"`
	NodeID    string               `json:"node_id"`
	Iteration int                  `json:"iteration"`
	Board     *graph.BoardSnapshot `json:"board"`
	Timestamp time.Time            `json:"timestamp"`
}

Checkpoint represents a snapshot of graph execution state.

type CheckpointCleaner ¶

type CheckpointCleaner interface {
	Cleanup(opts CleanupOptions) (deleted int, err error)
}

CheckpointCleaner supports periodic cleanup of old checkpoints.

type CheckpointManager ¶

type CheckpointManager interface {
	CheckpointStore
	List() ([]string, error)
	Delete(graphName string) error
}

CheckpointManager extends CheckpointStore with lifecycle operations. FileCheckpointStore implements this interface.

type CheckpointStore ¶

type CheckpointStore interface {
	Save(cp Checkpoint) error
	// Load retrieves the latest checkpoint. When runID is non-empty, only
	// checkpoints for that specific run are considered.
	Load(graphName, runID string) (*Checkpoint, error)
}

CheckpointStore is the interface for persisting and loading checkpoints.

type CleanupOptions ¶

type CleanupOptions struct {
	MaxAge   time.Duration
	MaxCount int
}

CleanupOptions configures checkpoint cleanup behavior.

type CloneableResolver ¶

type CloneableResolver interface {
	VariableResolver
	Clone() *variable.Resolver
}

CloneableResolver is an optional interface for resolvers that support cloning for use in parallel branches.

type Executor ¶

type Executor interface {
	Execute(ctx context.Context, g *graph.Graph, board *graph.Board, opts ...RunOption) (*graph.Board, error)
}

Executor is the interface for graph execution engines.

type FileCheckpointConfig ¶

type FileCheckpointConfig struct {
	Dir            string
	MaxCheckpoints int
}

FileCheckpointConfig configures the file-based checkpoint store.

type FileCheckpointStore ¶

type FileCheckpointStore struct {
	// contains filtered or unexported fields
}

FileCheckpointStore persists checkpoints as JSON files.

func NewFileCheckpointStore(cfg FileCheckpointConfig) (*FileCheckpointStore, error)

func (s *FileCheckpointStore) Cleanup(opts CleanupOptions) (int, error)

func (s *FileCheckpointStore) Delete(graphName string) error

func (s *FileCheckpointStore) List() ([]string, error)

func (s *FileCheckpointStore) Load(graphName, runID string) (*Checkpoint, error)

func (s *FileCheckpointStore) Save(cp Checkpoint) error

type LocalExecutor ¶

type LocalExecutor struct{}

LocalExecutor is the default single-process executor.

func NewLocalExecutor() *LocalExecutor

func (e *LocalExecutor) Execute(ctx context.Context, g *graph.Graph, board *graph.Board, opts ...RunOption) (*graph.Board, error)

type MergeFunc ¶

type MergeFunc func(board *graph.Board, snapshot *graph.BoardSnapshot, results []branchResult) error

MergeFunc merges parallel branch results into the parent board. snapshot is the pre-fork board state (for conflict detection).

type MergeStrategy ¶

type MergeStrategy string

MergeStrategy defines how parallel branch results are merged.

const (
	MergeLastWins        MergeStrategy = "last_wins"
	MergeNamespace       MergeStrategy = "namespace"
	MergeErrorOnConflict MergeStrategy = "error_on_conflict"
)

type ParallelConfig ¶

type ParallelConfig struct {
	Enabled       bool          `json:"enabled"`
	MaxBranches   int           `json:"max_branches"`
	MaxNesting    int           `json:"max_nesting"`
	MergeStrategy MergeStrategy `json:"merge_strategy"`
}

ParallelConfig configures parallel fork/join execution.

type RunOption ¶

type RunOption func(*runConfig)

RunOption configures a single graph execution run.

func WithCheckpointStore(s CheckpointStore) RunOption

func WithEventBus(bus event.Bus) RunOption

func WithMaxIterations(n int) RunOption

func WithMaxNodeRetries(n int) RunOption

func WithParallel(cfg ParallelConfig) RunOption

func WithResolver(r VariableResolver) RunOption

func WithRunID(id string) RunOption

func WithStartNode(id string) RunOption

func WithStreamCallback(cb graph.StreamCallback) RunOption

func WithTimeout(d time.Duration) RunOption

type Runner ¶

type Runner struct {
	// contains filtered or unexported fields
}

Runner is a lightweight, concurrency-safe graph executor.

It caches the CompiledGraph (static analysis result) and re-assembles fresh Node instances on every Run call, so concurrent callers never share mutable node state.

func NewRunner(def *graph.GraphDefinition, factory *node.Factory, opts ...RunnerOption) (*Runner, error)

func (r *Runner) Bus() event.Bus

func (r *Runner) Graph() (*graph.Graph, error)

func (r *Runner) Run(ctx context.Context, vars map[string]any, opts ...RunOption) (*graph.Board, error)

type RunnerOption ¶

type RunnerOption func(*Runner)

RunnerOption configures a Runner.

func WithRunnerEventBus(bus event.Bus) RunnerOption

func WithRunnerExecutor(e Executor) RunnerOption

type VariableResolver ¶

type VariableResolver interface {
	ResolveMap(m map[string]any) (map[string]any, error)
	AddScope(name string, vars map[string]any)
}

VariableResolver resolves variable references in node configs.